diff --git a/.github/workflows/deploy_terraform.yml b/.github/workflows/deploy_terraform.yml index 7ddbd9c8e..38e452512 100644 --- a/.github/workflows/deploy_terraform.yml +++ b/.github/workflows/deploy_terraform.yml @@ -538,7 +538,7 @@ jobs: # Deploy FastAPI Lambda # ============================================================ fast_api_lambda: - needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda, modelling_e2e_lambda] + needs: [determine_stage, ara_engine_lambda, categorisation_lambda, postcodeSplitter_lambda, bulk_address2uprn_combiner_lambda, bulkUploadFinaliser_lambda, modelling_e2e_lambda, bulk_document_download_lambda] uses: ./.github/workflows/_deploy_lambda.yml with: lambda_name: ara_fast_api @@ -816,6 +816,42 @@ jobs: TF_VAR_open_epc_api_token: ${{ secrets.DEV_OPEN_EPC_API_TOKEN }} TF_VAR_google_solar_api_key: ${{ secrets.DEV_GOOGLE_SOLAR_API_KEY }} + # ============================================================ + # Build Bulk Document Download image and Push + # ============================================================ + bulk_document_download_image: + needs: [determine_stage, shared_terraform] + uses: ./.github/workflows/_build_image.yml + with: + ecr_repo: bulk-document-download-${{ needs.determine_stage.outputs.stage }} + dockerfile_path: applications/bulk_document_download/Dockerfile + build_context: . + secrets: + AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }} + AWS_REGION: ${{ secrets.DEV_AWS_REGION }} + + # ============================================================ + # Deploy Bulk Document Download Lambda + # ============================================================ + bulk_document_download_lambda: + needs: [bulk_document_download_image, determine_stage] + uses: ./.github/workflows/_deploy_lambda.yml + with: + lambda_name: bulk_document_download + lambda_path: deployment/terraform/lambda/bulk_document_download + stage: ${{ needs.determine_stage.outputs.stage }} + ecr_repo: bulk-document-download-${{ needs.determine_stage.outputs.stage }} + image_digest: ${{ needs.bulk_document_download_image.outputs.image_digest }} + terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }} + secrets: + AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }} + AWS_REGION: ${{ secrets.DEV_AWS_REGION }} + TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }} + TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }} + TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }} + # ============================================================ # Deploy Hubspot ETL Lambda # ============================================================ diff --git a/CONTEXT.md b/CONTEXT.md index caaf3029e..c736d712d 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -99,6 +99,18 @@ _Avoid_: ventilation report, audit report, PAS ventilation (that is the external An externally-uploaded ventilation survey document produced by a human assessor and ingested from an external source (e.g. Coordination Hub). Recorded in `uploaded_files` with `file_type = PAS_2023_VENTILATION`. Distinct from a **Ventilation Audit**, which is machine-generated from MagicPlan floor plan data. _Avoid_: ventilation audit (that is the generated output) +**Document Type**: +The category of an uploaded document — the `uploaded_files.file_type` enum (photo_pack, site_note, the pas_2023_* / ecmk_* families, magic_plan_json, mcs_compliance_certificate, ventilation_audit, other, …). Nullable on the row. In a **Download Package** exactly one file per Document Type is included per property — the newest by `s3_upload_timestamp`; a row with a null Document Type is skipped and reported (ADR-0060). +_Avoid_: document category, file kind, doc class + +**Download Package**: +The single ZIP archive a **Bulk Document Download** produces: one folder per property (named by human-readable address, enriched from the hubspot deals data, with `landlord_property_id` appended for uniqueness), each folder holding the latest file of each **Document Type** held for that property. Built best-effort — properties/documents that don't resolve are skipped and listed in the run's `sub_task.outputs`, not failed (ADR-0060). Streamed to a dedicated `DOCUMENT_EXPORTS_BUCKET` (never held whole in memory — a package can be several GB); delivered as a 60-minute presigned URL. +_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 `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) — 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). 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. +_Avoid_: document export, bulk export, file dump + ### Source data **Site Notes**: diff --git a/applications/bulk_document_download/Dockerfile b/applications/bulk_document_download/Dockerfile new file mode 100644 index 000000000..e65d1bd9a --- /dev/null +++ b/applications/bulk_document_download/Dockerfile @@ -0,0 +1,35 @@ +FROM public.ecr.aws/lambda/python:3.11 + +# Postgres host/port/database are baked into the image at build time from the +# deploy workflow's --build-arg values (GitHub Actions DEV_DB_* secrets), +# mirroring the bulk_upload_finaliser Dockerfile. They map onto the POSTGRES_* +# names PostgresConfig.from_env reads. Username/password are NOT baked in -- +# Terraform injects those (and the SES SMTP credentials) as Lambda env vars from +# Secrets Manager. +ARG DEV_DB_HOST +ARG DEV_DB_PORT +ARG DEV_DB_NAME + +ENV POSTGRES_HOST=${DEV_DB_HOST} +ENV POSTGRES_PORT=${DEV_DB_PORT} +ENV POSTGRES_DATABASE=${DEV_DB_NAME} + +WORKDIR /var/task + +COPY applications/bulk_document_download/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# DDD-shaped packages only -- no pandas, no legacy backend/. +COPY datatypes/ datatypes/ +COPY domain/ domain/ +COPY infrastructure/ infrastructure/ +COPY orchestration/ orchestration/ +COPY repositories/ repositories/ +COPY utilities/ utilities/ +COPY applications/ applications/ + +# Place the handler at the Lambda task root so the runtime resolves +# ``main.handler`` without an extra package prefix. +COPY applications/bulk_document_download/handler.py /var/task/main.py + +CMD ["main.handler"] diff --git a/applications/bulk_document_download/bulk_document_download_trigger_body.py b/applications/bulk_document_download/bulk_document_download_trigger_body.py new file mode 100644 index 000000000..95da349b5 --- /dev/null +++ b/applications/bulk_document_download/bulk_document_download_trigger_body.py @@ -0,0 +1,19 @@ +from uuid import UUID + +from pydantic import BaseModel, ConfigDict + + +class BulkDocumentDownloadTriggerBody(BaseModel): + """SQS trigger for the bulk_document_download Lambda (ADR-0055/0060). + + Attach mode: the FastAPI route created the Task + SubTask and pinned the + recipe (``landlord_property_ids``, ``recipient_email``, ``package_name``) + onto the SubTask's ``inputs``. The message therefore carries only the + identifiers — the (potentially large) property set never travels through the + 256 KB SQS body. + """ + + model_config = ConfigDict(extra="allow") + + task_id: UUID + subtask_id: UUID diff --git a/applications/bulk_document_download/handler.py b/applications/bulk_document_download/handler.py new file mode 100644 index 000000000..a0d90798c --- /dev/null +++ b/applications/bulk_document_download/handler.py @@ -0,0 +1,100 @@ +"""bulk_document_download Lambda (ADR-0060). + +Attach-mode (ADR-0055) handler. The FastAPI route created the app-owned Task + +SubTask and pinned the recipe onto the SubTask's ``inputs``; this reads that +recipe, assembles the Download Package through the orchestrator, and returns the +result (presigned URL + skipped-document summary), which `TaskOrchestrator` +records on ``sub_task.outputs``. A wholly-empty selection raises +``SubTaskFailure`` from the orchestrator — recorded, not retried. + +PostgresConfig-only (POSTGRES_*), like the bulk_upload_finaliser Lambda. Source +documents are read from each row's own ``s3_file_bucket``; the ZIP is written to +the dedicated ``DOCUMENT_EXPORTS_BUCKET``. SES SMTP settings arrive as env vars +(terraform bakes the +Secrets Manager IAM-user credentials in). +""" + +from __future__ import annotations + +import os +from typing import Any + +import boto3 + +from applications.bulk_document_download.bulk_document_download_trigger_body import ( + BulkDocumentDownloadTriggerBody, +) +from domain.tasks.tasks import Source +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_downloader import S3DocumentDownloader +from orchestration.bulk_document_download_orchestrator import ( + BulkDocumentDownloadOrchestrator, +) +from repositories.property.landlord_address_resolver import LandlordAddressResolver +from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository +from repositories.uploaded_file.uploaded_file_postgres_repository import ( + UploadedFilePostgresRepository, +) +from utilities.aws_lambda.task_handler import task_handler + +# The Download Package link is valid for 60 minutes (ADR-0060). +_URL_TTL_SECONDS = 3600 + + +def _email_sender() -> SesSmtpEmailSender: + return SesSmtpEmailSender( + host=os.environ["SES_SMTP_HOST"], + port=int(os.environ["SES_SMTP_PORT"]), + username=os.environ["SES_SMTP_USERNAME"], + password=os.environ["SES_SMTP_PASSWORD"], + from_address=os.environ["SES_SMTP_FROM_ADDRESS"], + ) + + +@task_handler(task_source="bulk_document_download", source=Source.PORTFOLIO) +def handler(body: dict[str, Any], context: Any) -> dict[str, Any]: + trigger = BulkDocumentDownloadTriggerBody.model_validate(body) + engine = make_engine(PostgresConfig.from_env(os.environ)) + session = make_session(engine) + try: + subtask = SubTaskPostgresRepository(session).get(trigger.subtask_id) + recipe: dict[str, Any] = subtask.inputs or {} + landlord_property_ids: list[str] = [ + str(x) for x in recipe["landlord_property_ids"] + ] + recipient_email: str = str(recipe["recipient_email"]) + package_name: str = str(recipe["package_name"]) + + boto_s3: Any = boto3.client("s3") # pyright: ignore[reportUnknownMemberType] + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=UploadedFilePostgresRepository(session), + addresses=LandlordAddressResolver(session, landlord_property_ids), + documents=S3DocumentDownloader(boto_s3), + packages=S3Client(boto_s3, os.environ["DOCUMENT_EXPORTS_BUCKET"]), + email=_email_sender(), + url_ttl_seconds=_URL_TTL_SECONDS, + ) + result = orchestrator.run( + landlord_property_ids=landlord_property_ids, + recipient_email=recipient_email, + package_name=package_name, + ) + finally: + session.close() + + return { + "presigned_url": result.presigned_url, + "package_s3_key": result.package_s3_key, + "included": result.included, + "skipped": [ + { + "landlord_property_id": skip.landlord_property_id, + "s3_key": skip.s3_key, + "reason": skip.reason, + } + for skip in result.skipped + ], + } diff --git a/applications/bulk_document_download/requirements.txt b/applications/bulk_document_download/requirements.txt new file mode 100644 index 000000000..6a85a2552 --- /dev/null +++ b/applications/bulk_document_download/requirements.txt @@ -0,0 +1,4 @@ +boto3 +pydantic +sqlmodel +psycopg2-binary diff --git a/backend/app/config.py b/backend/app/config.py index af79b0106..8f0d551f1 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -46,6 +46,7 @@ class Settings(BaseSettings): LANDLORD_OVERRIDES_SQS_URL: str = "changeme" FINALISER_SQS_URL: str = "changeme" MODELLING_E2E_SQS_URL: str = "changeme" + BULK_DOCUMENT_DOWNLOAD_SQS_URL: str = "changeme" # Third parties EPC_AUTH_TOKEN: str = "changeme" diff --git a/backend/app/documents/__init__.py b/backend/app/documents/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/documents/download_tasks.py b/backend/app/documents/download_tasks.py new file mode 100644 index 000000000..7109224a6 --- /dev/null +++ b/backend/app/documents/download_tasks.py @@ -0,0 +1,122 @@ +"""The bulk-download route's view of the app-owned task's sub_task (ADR-0060). + +Parameterised SQL rather than the SQLModel mirrors: importing the +``infrastructure.postgres`` mirror of ``sub_task`` into the FastAPI process +double-registers the table and crashes at import (the same constraint the +Modelling Run distributor works around — see ``modelling/run_tasks.py``). +""" + +import json +from datetime import datetime, timezone +from typing import Any, cast +from uuid import UUID + +from sqlalchemy import text +from sqlmodel import Session + + +class DocumentDownloadTasks: + def __init__(self, session: Session) -> None: + self._session = session + + def read_selection_config(self, task_id: UUID) -> dict[str, Any]: + """The task's selection config, read from the FE-owned ``task.inputs`` + JSON (ADR-0060) — ``{portfolio_id, property_ids?, select_all?}``. Empty + dict when the task has no inputs.""" + row = ( + self._session.connection() + .execute( + text("SELECT inputs FROM tasks WHERE id = :task_id"), + {"task_id": task_id}, + ) + .first() + ) + if row is None or row[0] is None: + return {} + parsed: Any = json.loads(row[0]) + return cast(dict[str, Any], parsed) if isinstance(parsed, dict) else {} + + def already_distributed(self, task_id: UUID) -> bool: + """Whether the task already has a sub_task — the 409 guard against a + double-submit re-triggering the same download.""" + row = ( + self._session.connection() + .execute( + text("SELECT 1 FROM sub_task WHERE task_id = :task_id LIMIT 1"), + {"task_id": task_id}, + ) + .first() + ) + return row is not None + + def resolve_selection( + self, + project_codes: list[str], + property_ids: list[int], + ) -> list[str]: + """Resolve the selection to the distinct ``landlord_property_id`` set the + Download Package is built from (ADR-0060). Two independent sources, + unioned: + + - **project codes** — every property in the named HubSpot projects, read + from ``hubspot_deal_data`` (the project↔property grain lives there, not + on ``property``). + - **hand-picked property ids** — the selected ``property`` rows. + + Files are matched by ``landlord_property_id`` (the missing ``property_id`` + on ``uploaded_files`` is a known future gap), so a source row with a null + ``landlord_property_id`` contributes nothing and is dropped here. The + returned set size is the cap basis (property count, ADR-0060). + """ + landlord_property_ids: set[str] = set() + if project_codes: + rows = self._session.connection().execute( + text( + "SELECT DISTINCT landlord_property_id FROM hubspot_deal_data" + " WHERE project_code = ANY(:codes)" + " AND landlord_property_id IS NOT NULL" + ), + {"codes": project_codes}, + ) + landlord_property_ids.update(row[0] for row in rows.all()) + if property_ids: + rows = self._session.connection().execute( + text( + "SELECT landlord_property_id FROM property" + " WHERE id = ANY(:ids) AND landlord_property_id IS NOT NULL" + ), + {"ids": property_ids}, + ) + landlord_property_ids.update(row[0] for row in rows.all()) + return sorted(landlord_property_ids) + + def create_download_subtask( + self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any] + ) -> 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). + + 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) + result = self._session.connection().execute( + text( + "INSERT INTO sub_task (id, task_id, status, 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, + "task_id": task_id, + "inputs": json.dumps(recipe), + "updated_at": now, + }, + ) + self._session.commit() + return result.rowcount == 1 diff --git a/backend/app/documents/router.py b/backend/app/documents/router.py new file mode 100644 index 000000000..50b7a957b --- /dev/null +++ b/backend/app/documents/router.py @@ -0,0 +1,162 @@ +"""Bulk Document Download trigger: POST /v1/documents/bulk-download (ADR-0060). + +Resolves the authenticated requester's email, resolves + caps the property +selection, pins the recipe onto a single pre-created sub_task under the +app-owned task, and enqueues one message to the bulk_document_download worker. +Never assembles the package synchronously — the worker emails the link. +""" + +import json +from collections.abc import Callable, Iterator +from typing import Any, cast +from uuid import uuid4 + +import boto3 +from fastapi import APIRouter, Depends, HTTPException +from sqlmodel import Session + +from backend.app.config import get_settings +from backend.app.db.connection import db_engine +from backend.app.dependencies import validate_jwt_token, validate_token +from backend.app.documents.download_tasks import DocumentDownloadTasks +from backend.app.documents.schemas import BulkDownloadRequest + +# The selection cap (ADR-0060) — rejected at the trigger so an oversized request +# never reaches the worker. A conservative, tunable starting value. +MAX_PROPERTIES = 500 + +MessageSender = Callable[[str], None] + + +def get_session() -> Iterator[Session]: + with Session(db_engine) as session: + yield session + + +def get_message_sender() -> MessageSender: + settings = get_settings() + client: Any = cast(Any, boto3.client("sqs", settings.AWS_DEFAULT_REGION)) # pyright: ignore[reportUnknownMemberType] + queue_url = settings.BULK_DOCUMENT_DOWNLOAD_SQS_URL + + def send(body: str) -> None: + client.send_message(QueueUrl=queue_url, MessageBody=body) + + return send + + +def get_requesting_user_email(user: Any = Depends(validate_jwt_token)) -> str: + """The authenticated requester's email — the Download Package recipient. + + ADR-0059: the route resolves the user (JWT ``dbId`` -> ``UserModel``) rather + than only gating the token, and threads the email into the job.""" + 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, + detail="Could not determine the requesting user's email address.", + ) + return str(email) + + +router = APIRouter( + prefix="/documents", + tags=["documents"], + dependencies=[Depends(validate_token)], +) + + +@router.post("/bulk-download", status_code=202) +async def bulk_download( + body: BulkDownloadRequest, + session: Session = Depends(get_session), + send_message: MessageSender = Depends(get_message_sender), + recipient_email: str = Depends(get_requesting_user_email), +) -> dict[str, str]: + tasks = DocumentDownloadTasks(session) + if tasks.already_distributed(body.task_id): + raise HTTPException( + status_code=409, + detail=( + f"Task {body.task_id} already has a sub_task — a download has " + "already been started for it." + ), + ) + + config = tasks.read_selection_config(body.task_id) + portfolio_id: Any = config.get("portfolio_id") + raw_project_codes: Any = config.get("project_codes") or [] + raw_property_ids: Any = config.get("property_ids") or [] + if not isinstance(raw_project_codes, list) or not all( + isinstance(code, str) for code in cast(list[Any], raw_project_codes) + ): + raise HTTPException( + status_code=400, + detail="task.inputs project_codes must be a list of strings.", + ) + if not isinstance(raw_property_ids, list) or not all( + isinstance(pid, int) for pid in cast(list[Any], raw_property_ids) + ): + raise HTTPException( + status_code=400, + detail="task.inputs property_ids must be a list of integers.", + ) + project_codes: list[str] = cast(list[str], raw_project_codes) + property_ids: list[int] = cast(list[int], raw_property_ids) + if not project_codes and not property_ids: + raise HTTPException( + status_code=400, + detail="task.inputs must provide project_codes or property_ids.", + ) + + landlord_property_ids = tasks.resolve_selection(project_codes, property_ids) + if not landlord_property_ids: + raise HTTPException( + status_code=400, + detail=( + "The selection resolves to no downloadable properties — none of " + "the selected projects or properties has a landlord_property_id." + ), + ) + if len(landlord_property_ids) > MAX_PROPERTIES: + raise HTTPException( + status_code=400, + detail=( + f"The selection has {len(landlord_property_ids)} properties, over " + f"the {MAX_PROPERTIES} limit — narrow your selection." + ), + ) + + subtask_id = uuid4() + package_name = ( + f"portfolio-{portfolio_id}-{body.task_id}" + if isinstance(portfolio_id, int) + else f"documents-{body.task_id}" + ) + recipe = { + "landlord_property_ids": landlord_property_ids, + "recipient_email": recipient_email, + "package_name": package_name, + } + 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)}) + ) + return {"message": "Bulk document download started"} diff --git a/backend/app/documents/schemas.py b/backend/app/documents/schemas.py new file mode 100644 index 000000000..c383ddfaf --- /dev/null +++ b/backend/app/documents/schemas.py @@ -0,0 +1,20 @@ +from uuid import UUID + +from pydantic import BaseModel + + +class BulkDownloadRequest(BaseModel): + """A request to assemble a Download Package (ADR-0060). + + The front end creates the app-owned Task first and writes the **selection + config** into ``task.inputs`` (a JSON object of + ``{project_codes?: str[], property_ids?: int[], portfolio_id?: int}`` — so a + large selection never travels in an HTTP body), then passes only the + ``task_id`` here. The backend reads the selection from ``task.inputs`` and + resolves it to the distinct ``landlord_property_id`` set: the union of every + property in the named HubSpot ``project_codes`` (from ``hubspot_deal_data``) + and the hand-picked ``property_ids``. At least one of the two must be given; + ``portfolio_id`` is optional and used only to name the package. + """ + + task_id: UUID diff --git a/backend/app/main.py b/backend/app/main.py index 4b4aaaa70..270917f63 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -11,6 +11,7 @@ from backend.app.whlg import router as whlg_router from backend.app.plan import router as plan_router from backend.app.tasks import router as tasks_router from backend.app.bulk_uploads import router as bulk_uploads_router +from backend.app.documents import router as documents_router from backend.app.modelling import router as modelling_router from backend.app.dependencies import validate_api_key from backend.app.config import get_settings @@ -65,6 +66,7 @@ app.include_router(plan_router.router, prefix="/v1") app.include_router(whlg_router.router, prefix="/v1") app.include_router(bulk_uploads_router.router, prefix="/v1") app.include_router(modelling_router.router, prefix="/v1") +app.include_router(documents_router.router, prefix="/v1") if get_settings().ENVIRONMENT == "local": from backend.app.local import router as local_router diff --git a/deployment/terraform/lambda/bulk_document_download/main.tf b/deployment/terraform/lambda/bulk_document_download/main.tf new file mode 100644 index 000000000..3325a1ff1 --- /dev/null +++ b/deployment/terraform/lambda/bulk_document_download/main.tf @@ -0,0 +1,111 @@ +############################################ +# Credentials from Secrets Manager (resolved at plan/apply time and baked into +# env vars). The SES SMTP credentials are an IAM *user* (see modules/ses), so +# the Lambda authenticates to SES over SMTP with them — the execution role needs +# neither ses:* nor runtime secretsmanager access. +############################################ +data "aws_secretsmanager_secret_version" "db_credentials" { + secret_id = "${var.stage}/assessment_model/db_credentials" +} + +data "aws_secretsmanager_secret_version" "ses_smtp" { + secret_id = "${var.stage}/ses/smtp_credentials" +} + +locals { + db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string) + ses_smtp = jsondecode(data.aws_secretsmanager_secret_version.ses_smtp.secret_string) + + document_exports_bucket = "retrofit-document-exports-${var.stage}" + + # Source buckets the uploaded_files rows point at (read for packaging). + # OPEN QUESTION: confirm this is the closed set — if files can live in + # arbitrary buckets, widen to ["arn:aws:s3:::*"]. + source_bucket_arns = [ + "arn:aws:s3:::retrofit-data-${var.stage}", + "arn:aws:s3:::retrofit-energy-assessments-${var.stage}", + ] +} + +############################################ +# Lambda + SQS queue + trigger +############################################ +module "lambda" { + source = "../../modules/lambda_with_sqs" + + name = var.lambda_name + stage = var.stage + + image_uri = local.image_uri + + reserved_concurrent_executions = var.reserved_concurrent_executions + + batch_size = var.batch_size + maximum_concurrency = var.maximum_concurrency + + timeout = 900 + memory_size = 3008 + # A Download Package can be several GB; it is streamed to /tmp before the + # multipart upload (ADR-0060), so raise ephemeral storage to the 10 GB max. + ephemeral_storage_size = 10240 + + environment = { + STAGE = var.stage + LOG_LEVEL = "info" + + POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username + POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password + POSTGRES_HOST = var.db_host + POSTGRES_DATABASE = var.db_name + POSTGRES_PORT = var.db_port + + DOCUMENT_EXPORTS_BUCKET = local.document_exports_bucket + + # SES SMTP — IAM-user credentials sourced from Secrets Manager (modules/ses). + SES_SMTP_HOST = "email-smtp.eu-west-2.amazonaws.com" + SES_SMTP_PORT = "587" + SES_SMTP_USERNAME = local.ses_smtp.username + SES_SMTP_PASSWORD = local.ses_smtp.password + SES_SMTP_FROM_ADDRESS = var.ses_from_address + } +} + +############################################ +# IAM: read uploaded documents from their source buckets +############################################ +module "s3_read" { + source = "../../modules/s3_iam_policy" + + policy_name = "BulkDocumentDownloadS3Read-${var.stage}" + policy_description = "Allow bulk_document_download Lambda to read uploaded_files objects from their source buckets" + bucket_arns = local.source_bucket_arns + actions = ["s3:GetObject", "s3:ListBucket"] + resource_paths = ["/*"] +} + +resource "aws_iam_role_policy_attachment" "s3_read" { + role = module.lambda.role_name + policy_arn = module.s3_read.policy_arn +} + +############################################ +# IAM: write + read the assembled ZIP on the dedicated exports bucket. +# GetObject is required so the Lambda-signed presigned GET URL resolves. +############################################ +module "s3_exports" { + source = "../../modules/s3_iam_policy" + + policy_name = "BulkDocumentDownloadS3Exports-${var.stage}" + policy_description = "Allow bulk_document_download Lambda to write and presign-read Download Packages on the exports bucket" + bucket_arns = ["arn:aws:s3:::${local.document_exports_bucket}"] + actions = ["s3:PutObject", "s3:GetObject"] + resource_paths = ["/*"] +} + +resource "aws_iam_role_policy_attachment" "s3_exports" { + role = module.lambda.role_name + policy_arn = module.s3_exports.policy_arn +} + +# NOTE: no ses:* on the role — the handler sends over SMTP with the SES IAM-user +# credentials injected above. diff --git a/deployment/terraform/lambda/bulk_document_download/outputs.tf b/deployment/terraform/lambda/bulk_document_download/outputs.tf new file mode 100644 index 000000000..7f2442c26 --- /dev/null +++ b/deployment/terraform/lambda/bulk_document_download/outputs.tf @@ -0,0 +1,9 @@ +output "bulk_document_download_queue_url" { + value = module.lambda.queue_url + description = "URL of the bulk-document-download SQS queue (FastAPI enqueues jobs here)" +} + +output "bulk_document_download_queue_arn" { + value = module.lambda.queue_arn + description = "ARN of the bulk-document-download SQS queue" +} diff --git a/deployment/terraform/lambda/bulk_document_download/provider.tf b/deployment/terraform/lambda/bulk_document_download/provider.tf new file mode 100644 index 000000000..2ea6c350c --- /dev/null +++ b/deployment/terraform/lambda/bulk_document_download/provider.tf @@ -0,0 +1,20 @@ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = ">= 5.0" + } + } + + backend "s3" { + bucket = "bulk-document-download-terraform-state" + key = "terraform.tfstate" + region = "eu-west-2" + } + + required_version = ">= 1.2.0" +} + +provider "aws" { + region = "eu-west-2" +} diff --git a/deployment/terraform/lambda/bulk_document_download/variables.tf b/deployment/terraform/lambda/bulk_document_download/variables.tf new file mode 100644 index 000000000..9abb25362 --- /dev/null +++ b/deployment/terraform/lambda/bulk_document_download/variables.tf @@ -0,0 +1,61 @@ +variable "lambda_name" { + type = string + description = "Logical name of the lambda" +} + +variable "stage" { + description = "Deployment stage (e.g. dev, prod)" + type = string +} + +variable "ecr_repo_url" { + type = string + description = "ECR repository URL (no tag, no digest)" +} + +variable "image_digest" { + type = string + description = "Image digest (sha256:...)" +} + +variable "reserved_concurrent_executions" { + type = number + default = -1 + description = "Reserved concurrency for the Lambda. -1 = unreserved." +} + +variable "maximum_concurrency" { + type = number + default = 5 + description = "Maximum concurrent Lambda invocations from the SQS trigger." +} + +variable "batch_size" { + type = number + default = 1 +} + +variable "db_host" { + type = string + sensitive = true +} + +variable "db_name" { + type = string + sensitive = true +} + +variable "db_port" { + type = string + sensitive = true +} + +variable "ses_from_address" { + type = string + description = "Verified SES sender for Download Package notifications (ADR-0059)." + default = "noreply@domna.homes" +} + +locals { + image_uri = "${var.ecr_repo_url}@${var.image_digest}" +} diff --git a/deployment/terraform/lambda/fast-api/main.tf b/deployment/terraform/lambda/fast-api/main.tf index 26e154d6a..73310b999 100644 --- a/deployment/terraform/lambda/fast-api/main.tf +++ b/deployment/terraform/lambda/fast-api/main.tf @@ -73,6 +73,15 @@ data "terraform_remote_state" "modelling_e2e" { } } +data "terraform_remote_state" "bulk_document_download" { + backend = "s3" + config = { + bucket = "bulk-document-download-terraform-state", + key = "env:/${var.stage}/terraform.tfstate" + region = "eu-west-2" + } +} + ############################################ # Load Credentials ############################################ @@ -136,6 +145,8 @@ module "fastapi" { FINALISER_SQS_URL = data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_url LANDLORD_OVERRIDES_SQS_URL = data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_url MODELLING_E2E_SQS_URL = data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_url + + BULK_DOCUMENT_DOWNLOAD_SQS_URL = data.terraform_remote_state.bulk_document_download.outputs.bulk_document_download_queue_url } } @@ -160,7 +171,8 @@ module "fastapi_sqs_policy" { data.terraform_remote_state.bulk_address2uprn_combiner.outputs.bulk_address2uprn_combiner_queue_arn, data.terraform_remote_state.bulk_upload_finaliser.outputs.bulk_upload_finaliser_queue_arn, data.terraform_remote_state.landlord_description_overrides.outputs.landlord_description_overrides_queue_arn, - data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_arn + data.terraform_remote_state.modelling_e2e.outputs.modelling_e2e_queue_arn, + data.terraform_remote_state.bulk_document_download.outputs.bulk_document_download_queue_arn ] conditions = null diff --git a/deployment/terraform/modules/lambda_service/main.tf b/deployment/terraform/modules/lambda_service/main.tf index 3250110ba..0981deac1 100644 --- a/deployment/terraform/modules/lambda_service/main.tf +++ b/deployment/terraform/modules/lambda_service/main.tf @@ -11,6 +11,10 @@ resource "aws_lambda_function" "this" { reserved_concurrent_executions = var.reserved_concurrent_executions + ephemeral_storage { + size = var.ephemeral_storage_size + } + environment { variables = var.environment } diff --git a/deployment/terraform/modules/lambda_service/variables.tf b/deployment/terraform/modules/lambda_service/variables.tf index 46241f300..c0ddc10ff 100644 --- a/deployment/terraform/modules/lambda_service/variables.tf +++ b/deployment/terraform/modules/lambda_service/variables.tf @@ -12,6 +12,12 @@ variable "memory_size" { default = 512 } +variable "ephemeral_storage_size" { + type = number + default = 512 + description = "Lambda /tmp size in MB (512-10240). Default 512 keeps every existing Lambda unchanged; raised only where a large on-disk artifact is built (e.g. multi-GB Download Package ZIPs, ADR-0060)." +} + variable "environment" { type = map(string) default = {} diff --git a/deployment/terraform/modules/lambda_with_sqs/main.tf b/deployment/terraform/modules/lambda_with_sqs/main.tf index 97f867935..2d8259dcf 100644 --- a/deployment/terraform/modules/lambda_with_sqs/main.tf +++ b/deployment/terraform/modules/lambda_with_sqs/main.tf @@ -31,6 +31,8 @@ module "lambda" { timeout = var.timeout memory_size = var.memory_size + ephemeral_storage_size = var.ephemeral_storage_size + environment = var.environment reserved_concurrent_executions = var.reserved_concurrent_executions } diff --git a/deployment/terraform/modules/lambda_with_sqs/variables.tf b/deployment/terraform/modules/lambda_with_sqs/variables.tf index 90585e929..d75ff28e8 100644 --- a/deployment/terraform/modules/lambda_with_sqs/variables.tf +++ b/deployment/terraform/modules/lambda_with_sqs/variables.tf @@ -25,6 +25,11 @@ variable "memory_size" { default = 1024 } +variable "ephemeral_storage_size" { + type = number + default = 512 +} + variable "environment" { type = map(string) default = {} diff --git a/deployment/terraform/shared/main.tf b/deployment/terraform/shared/main.tf index b4cdafbcc..a492401de 100644 --- a/deployment/terraform/shared/main.tf +++ b/deployment/terraform/shared/main.tf @@ -895,6 +895,38 @@ output "modelling_e2e_ecr_url" { } +################################################ +# Bulk Document Download – Lambda (ADR-0060) +################################################ +module "bulk_document_download_state_bucket" { + source = "../modules/tf_state_bucket" + bucket_name = "bulk-document-download-terraform-state" +} + +module "bulk_document_download_registry" { + source = "../modules/container_registry" + name = "bulk-document-download" + stage = var.stage +} + +# Dedicated bucket for generated Download Packages — kept out of the shared data +# bucket so retention/IAM stay scoped to exports (ADR-0060). +module "document_exports" { + source = "../modules/s3" + bucketname = "retrofit-document-exports-${var.stage}" + allowed_origins = var.allowed_origins +} + +output "document_exports_bucket_name" { + value = module.document_exports.bucket_name + description = "Name of the document exports bucket (Download Packages)" +} + +output "bulk_document_download_ecr_url" { + value = module.bulk_document_download_registry.repository_url +} + + ################################################ # Abri OpenHousing – Lambda ################################################ diff --git a/docs/adr/0059-outbound-email-via-ses-smtp-adapter.md b/docs/adr/0059-outbound-email-via-ses-smtp-adapter.md new file mode 100644 index 000000000..fb2946eae --- /dev/null +++ b/docs/adr/0059-outbound-email-via-ses-smtp-adapter.md @@ -0,0 +1,79 @@ +--- +status: accepted +--- + +# Outbound email is sent from the backend via an SES-SMTP adapter, with the recipient threaded from the trigger route + +The Bulk Document Download feature (ADR-0060) must email a completed package's +link to the user who requested it. This is the **first outbound email in the +Python backend** — there is no `smtplib`/SES/SendGrid code anywhere today. Two +facts force a decision rather than a copy: + +1. **Where the email is sent from.** The front end can send email, but the + package is produced **asynchronously in a Lambda**, long after the trigger + request has returned `202`. The completion event exists only on the backend. +2. **Who to email.** Auth today only *gates* the FastAPI routes + (`validate_token` returns the raw token, not the user); neither the handler + nor the `Task`/`SubTask` model captures a user. The requesting identity is + provable (the NextAuth JWT carries `dbId` → `UserModel.email`) but is not + currently threaded anywhere. + +## Decision + +**The backend sends the email, through a DDD port + SES-SMTP adapter, to a +recipient the trigger route resolves and pins into `tasks.inputs`.** + +- **Send from the backend, on completion.** The job that finishes the work + owns the notification — the orchestrator calls an injected email port after + the package is uploaded and the URL minted. The front end is not asked to + poll-then-send. +- **Port + adapter (hexagonal).** A `repositories/email/` port + (`EmailSender` — `send(to, subject, body/html)`) with an + `infrastructure/email/ses_smtp_email_sender.py` adapter over `smtplib`, using + the **existing SES SMTP** setup (terraform `modules/ses`, domain + `domna.homes`, credentials already in Secrets Manager + `${stage}/ses/smtp_credentials`). SMTP, not the SES SendEmail API, because + that is what the provisioned infrastructure exposes. The port is injected into + the orchestrator so the send is testable behind an interface and the transport + is swappable. +- **Recipient threaded at the route, pinned in `inputs`.** The FastAPI trigger + route resolves the authenticated user (`dbId` → `UserModel.email`, via the + existing `get_user`) and writes the email into the `tasks.inputs` JSON at task + creation. The Lambda reads it from `inputs` and never touches auth. This + graduates the route from *gating* the token to *injecting* the user — a + deliberate, contained change to `backend/app/dependencies.py` usage at the one + new route. +- **Config follows the Lambda convention.** SMTP host/username/password reach + the Lambda as environment variables (sourced from the Secrets Manager entry in + terraform), read via `os.environ`/`PostgresConfig`-style config — not + `backend/app/config.py get_settings()`, which is the legacy app boundary. + +## Considered options + +- **Front end sends the email.** Rejected: the completion moment lives in the + Lambda; making the FE responsible means it must poll task status to + completion and then send — more moving parts, and the notification would lag + or be missed if the user closes the tab. Backend-sends keeps the trigger and + the notification on the same side of the queue. +- **SES SendEmail API (boto3 `ses`) instead of SMTP.** Rejected for now: the + provisioned infrastructure issues SMTP credentials; using the API would need + new IAM + identity wiring. Revisit if a templated/bulk API becomes worthwhile. +- **Resolve the recipient in the Lambda from `portfolio_id` → `PortfolioUsers`.** + Rejected: a portfolio has many users, so the join cannot identify *the + requester* — real risk of emailing the wrong person or several. The requester + is only unambiguous at the authenticated route. +- **FE passes the recipient email in the trigger body.** Rejected: trusts the + client for an identity the JWT already proves, and can drift from the + logged-in user. + +## Consequences + +- A reusable `EmailSender` capability now exists for future backend + notifications (not just this feature). +- The one new FastAPI route must inject the user (not merely gate) and persist + the email in `inputs`; `Task`/`SubTask` still gain no user column — the + recipient lives in the request payload, scoped to the job. +- Email content is plain for v1 (a link, a 60-minute expiry note, and the + skipped-document summary from `sub_task.outputs`); templating can follow. +- A new terraform env-var surface (SMTP creds from the existing Secrets Manager + entry) is added to the Lambda; no new SES infrastructure is required. diff --git a/docs/adr/0060-bulk-document-download-package-model.md b/docs/adr/0060-bulk-document-download-package-model.md new file mode 100644 index 000000000..10959d19a --- /dev/null +++ b/docs/adr/0060-bulk-document-download-package-model.md @@ -0,0 +1,118 @@ +--- +status: accepted (builds on ADR-0055, ADR-0059) +--- + +# Bulk Document Download builds one capped, best-effort Download Package per request + +Users need to pull the documents held in `uploaded_files` for many properties +at once — per property, the latest file of each **Document Type** — as a single +archive, without clicking through the UI file by file. The request is initiated +in the front end, can span a hand-picked set of properties or one or more whole +HubSpot projects (by project code), and finishes with a link emailed to the +requester (ADR-0059). + +`uploaded_files` links to a property by `uprn` **or** `landlord_property_id` +(both nullable) and has **no `property_id`** column; `file_type` (the Document +Type) is itself nullable; and files live across arbitrary buckets +(`s3_file_bucket` per row). + +## Decision + +**A request produces exactly one Download Package: a ZIP of one folder per +property, each holding the latest file of each Document Type — built +best-effort, size-capped at the trigger, on the app-owned-task + attach-mode +lane (ADR-0055).** + +- **Trigger & lifecycle (ADR-0055).** The **front end** creates the app-owned + `tasks` row and writes the **selection config** — + `{project_codes?: str[], property_ids?: int[], portfolio_id?: int}` — into + `tasks.inputs` (TEXT/JSON, FE-owned), then calls the FastAPI route with **only + the `task_id`**. A large selection therefore never travels in an HTTP body. + The route **reads the selection from `tasks.inputs`** and resolves it to the + distinct `landlord_property_id` set — the **union** of every property in the + named HubSpot `project_codes` (read from `hubspot_deal_data`, where the + project↔property grain lives — `property` carries only `portfolio_id`) and the + hand-picked `property_ids`. At least one of the two must be provided; + `portfolio_id` is optional and used only to name the package. The route then + caps the set, resolves the recipient email from the authenticated user + (ADR-0059), **pins the resolved recipe** + (`landlord_property_ids`, `recipient_email`, `package_name`) onto **one + pre-created `sub_task`'s `inputs`**, and drops one SQS message (`task_id`, + `sub_task_id`). The new `applications/bulk_document_download` Lambda runs in + **attach mode**, reading that recipe from the sub_task; `TaskOrchestrator` + owns status + roll-up. (`task.inputs` = the FE's selection; `sub_task.inputs` + = the backend's resolved recipe.) +- **One package = one sub_task, streamed.** No fan-out. A Download Package can + be **several GB**, so it is never held whole in memory: each document is read + from its own `s3_file_bucket`, written into an on-disk ZIP in `/tmp`, and + released; the finished archive is **multipart-uploaded from disk** + (`S3Client.upload_file`). The Lambda's ephemeral storage (`/tmp`) is raised + accordingly (up to 10 GB) — see Consequences. One artifact, one URL, one + email. +- **Selection cap at the route.** Reject `> N` properties synchronously with a + legible "narrow your selection" error (property count, not a document/byte + 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/`). +- **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`. + Rows with a **null Document Type are skipped** and listed in the run output. +- **Best-effort failure contract.** Build from whatever resolves. Skipped + properties (no documents) and skipped null-type files are recorded in + `sub_task.outputs`. The run **fails only** on an infrastructure error + (S3/DB/zip/upload/email) or when the whole selection yields **zero + documents**. The email carries an "N properties, M documents, X skipped" + summary. +- **Delivery — both channels.** The presigned URL (60-minute expiry) and the + skip-summary are written to `sub_task.outputs` (the FE already polls task + status and can show the link) **and** emailed (ADR-0059). +- **A dedicated exports bucket.** Packages are written to a **separate + `DOCUMENT_EXPORTS_BUCKET`**, not `DATA_BUCKET` — the generated archives are a + distinct class of artifact (transient, user-facing, wide-read source but + single-writer) and keeping them out of the shared data bucket keeps IAM and + any future retention policy cleanly scoped to exports. No lifecycle rule is + imposed on `DATA_BUCKET`. Retention on the exports bucket is an open decision + (the packages are transient, so an expiry is likely warranted, but it is left + to a follow-up rather than baked in here). + +## Considered options + +- **Fan out into per-batch sub_tasks / multiple ZIPs** (like the modelling + run). Rejected: the user would receive N links / N emails and "the download" + would stop being one file; the capped-single-package model keeps the artifact + and the notification singular. +- **Cap on resolved document count / bytes.** Rejected for v1: it adds + synchronous DB + S3 HEAD work to the trigger and couples the route to the + packaging logic; a property-count cap is cheap and predictable. +- **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. + +## 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 + `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 + portfolio changes between trigger and execution. +- `N`, the Lambda's ephemeral-storage (`/tmp`) size, and its memory are tunable + knobs, not load-bearing invariants. `/tmp` must be raised beyond the 512 MB + default (up to 10 GB) to hold a multi-GB archive — a new terraform knob on the + shared Lambda modules, backward-compatible (every other Lambda keeps 512 MB). +- 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. diff --git a/domain/bulk_document_download/__init__.py b/domain/bulk_document_download/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/domain/bulk_document_download/package_plan.py b/domain/bulk_document_download/package_plan.py new file mode 100644 index 000000000..ba12e3560 --- /dev/null +++ b/domain/bulk_document_download/package_plan.py @@ -0,0 +1,104 @@ +"""The Download Package plan (ADR-0060). + +Pure logic: given the documents resolved for a set of properties, decide what +goes where in the ZIP — one folder per property, the latest file of each +Document Type — and which rows are skipped and reported. No I/O: the +orchestrator does the S3/DB work; this module only lays out the archive. +""" + +from __future__ import annotations + +import posixpath +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, Sequence + + +@dataclass(frozen=True) +class ResolvedDocument: + """One `uploaded_files` row resolved for packaging: its property identity + (matched by `landlord_property_id`, ADR-0060), the display address enriched + from the hubspot deals data, its Document Type (`file_type`, nullable), the + S3 location to read it from, and when it was uploaded (for latest-per-type). + """ + + landlord_property_id: str + address: str + document_type: Optional[str] + s3_bucket: str + s3_key: str + uploaded_at: datetime + + +@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. 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) +class SkippedDocument: + """A resolved row left out of the package, with the reason (surfaced in the + run's `sub_task.outputs`, ADR-0060).""" + + landlord_property_id: str + s3_key: str + reason: str + + +@dataclass(frozen=True) +class DownloadPackagePlan: + """The laid-out archive: the files to include and the rows skipped.""" + + entries: tuple[PackageEntry, ...] + skipped: tuple[SkippedDocument, ...] + + +def _extension_of(s3_key: str) -> str: + """The file extension of an S3 key (``.pdf``, ``.zip``, …), empty when the + key's basename has none. Used to keep the packaged file openable while + naming it by Document Type.""" + _root, ext = posixpath.splitext(posixpath.basename(s3_key)) + return ext + + +def plan_download_package( + documents: Sequence[ResolvedDocument], +) -> DownloadPackagePlan: + """Lay out a Download Package from resolved documents (ADR-0060): one folder + per property, the latest file of each Document Type, null-Document-Type rows + skipped and reported.""" + latest_by_group: dict[tuple[str, str], ResolvedDocument] = {} + skipped: list[SkippedDocument] = [] + for document in documents: + if document.document_type is None: + skipped.append( + SkippedDocument( + landlord_property_id=document.landlord_property_id, + s3_key=document.s3_key, + reason="null_document_type", + ) + ) + continue + group = (document.landlord_property_id, document.document_type) + incumbent = latest_by_group.get(group) + if incumbent is None or document.uploaded_at > incumbent.uploaded_at: + latest_by_group[group] = document + + entries: list[PackageEntry] = [ + PackageEntry( + zip_path=f"{document.address} ({document.landlord_property_id})/" + 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() + ] + return DownloadPackagePlan(entries=tuple(entries), skipped=tuple(skipped)) diff --git a/infrastructure/email/__init__.py b/infrastructure/email/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/infrastructure/email/ses_smtp_email_sender.py b/infrastructure/email/ses_smtp_email_sender.py new file mode 100644 index 000000000..ba53d23bc --- /dev/null +++ b/infrastructure/email/ses_smtp_email_sender.py @@ -0,0 +1,71 @@ +"""SES-over-SMTP implementation of the outbound-email port (ADR-0059). + +Sends via `smtplib` against the SES SMTP endpoint using the credentials +provisioned in terraform (`modules/ses`, Secrets Manager +`${stage}/ses/smtp_credentials`). The SMTP transport is injected so the sender +is testable without a live server. +""" + +from __future__ import annotations + +import smtplib +from email.message import EmailMessage +from types import TracebackType +from typing import Any, Callable, Optional, Protocol, Type, cast + + +class SmtpTransport(Protocol): + """The minimal SMTP surface the sender uses — satisfied structurally by + `smtplib.SMTP` and by test fakes.""" + + def __enter__(self) -> "SmtpTransport": ... + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + tb: Optional[TracebackType], + ) -> Optional[bool]: ... + + def starttls(self) -> Any: ... + + def login(self, user: str, password: str) -> Any: ... + + def send_message(self, message: EmailMessage) -> Any: ... + + +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)) + + +class SesSmtpEmailSender: + def __init__( + self, + *, + host: str, + port: int, + username: str, + password: str, + from_address: str, + smtp_factory: Callable[[str, int], SmtpTransport] = _default_smtp, + ) -> None: + self._host = host + self._port = port + self._username = username + self._password = password + self._from_address = from_address + self._smtp_factory = smtp_factory + + def send(self, *, to: str, subject: str, body: str) -> None: + message = EmailMessage() + message["From"] = self._from_address + message["To"] = to + message["Subject"] = subject + message.set_content(body) + with self._smtp_factory(self._host, self._port) as smtp: + smtp.starttls() + smtp.login(self._username, self._password) + smtp.send_message(message) diff --git a/infrastructure/postgres/task_table.py b/infrastructure/postgres/task_table.py index 32e5450bb..4ee46dc4d 100644 --- a/infrastructure/postgres/task_table.py +++ b/infrastructure/postgres/task_table.py @@ -2,7 +2,7 @@ from datetime import datetime, timezone from typing import ClassVar, Optional from uuid import UUID, uuid4 -from sqlalchemy import Column +from sqlalchemy import Column, Text from sqlalchemy import Enum as SAEnum from sqlmodel import Field, SQLModel @@ -18,6 +18,11 @@ class TaskRow(SQLModel, table=True): job_completed: Optional[datetime] = None status: str = Field(default="waiting") service: Optional[str] = None + # FE-owned (Drizzle) TEXT column holding the task's request as a JSON + # string. Declared here so the backend can read it (e.g. the Bulk Document + # Download route reads its property selection from it, ADR-0060) and so the + # test schema builds the column; prod owns the real column via FE migrations. + inputs: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True)) updated_at: datetime = Field( default_factory=lambda: datetime.now(timezone.utc) ) diff --git a/infrastructure/s3/s3_client.py b/infrastructure/s3/s3_client.py index a789fcc22..5d5dc55ae 100644 --- a/infrastructure/s3/s3_client.py +++ b/infrastructure/s3/s3_client.py @@ -20,3 +20,21 @@ class S3Client: def put_object(self, key: str, body: bytes) -> str: self._client.put_object(Bucket=self._bucket, Key=key, Body=body) return f"s3://{self._bucket}/{key}" + + def upload_file(self, local_path: str, key: str) -> str: + """Upload a file from local disk, using boto's managed transfer so a + large object (a multi-GB Download Package ZIP, ADR-0060) is streamed in + multipart from ``/tmp`` rather than held in memory.""" + self._client.upload_file(Filename=local_path, Bucket=self._bucket, Key=key) + return f"s3://{self._bucket}/{key}" + + def generate_presigned_url(self, key: str, expires_in: int) -> str: + """A time-limited URL that lets the holder GET this object without AWS + credentials (ADR-0060 — how a Download Package link is delivered). + ``expires_in`` is the validity window in seconds.""" + url: str = self._client.generate_presigned_url( + "get_object", + Params={"Bucket": self._bucket, "Key": key}, + ExpiresIn=expires_in, + ) + return url 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 new file mode 100644 index 000000000..c3c0db2cd --- /dev/null +++ b/orchestration/bulk_document_download_orchestrator.py @@ -0,0 +1,176 @@ +"""The Bulk Document Download orchestrator (ADR-0060). + +Ties the pieces together for one run: gather a property set's uploaded files +(by `landlord_property_id`), enrich each with its display address, lay them out +with the Download Package plan, stream the chosen files into a single ZIP, +upload it, mint a presigned URL, and email the requester. Best-effort — skipped +rows are reported, not fatal; a wholly-empty selection is a recorded failure +(ADR-0060). +""" + +from __future__ import annotations + +import os +import tempfile +import zipfile +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, Protocol, Sequence, cast + +from domain.bulk_document_download.package_plan import ( + ResolvedDocument, + SkippedDocument, + plan_download_package, +) +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 + + +class UploadedFileReader(Protocol): + def by_landlord_property_ids( + self, landlord_property_ids: Sequence[str] + ) -> list[UploadedFile]: ... + + +class AddressResolver(Protocol): + def address_for(self, landlord_property_id: str) -> str: ... + + +class DocumentDownloader(Protocol): + def download(self, bucket: str, key: str, dest_path: str) -> None: ... + + +@dataclass(frozen=True) +class DownloadPackageResult: + """What a completed run reports (lands in `sub_task.outputs`, ADR-0060).""" + + presigned_url: str + package_s3_key: str + included: int + skipped: tuple[SkippedDocument, ...] + + +class BulkDocumentDownloadOrchestrator: + def __init__( + self, + *, + uploaded_files: UploadedFileReader, + addresses: AddressResolver, + documents: DocumentDownloader, + packages: S3Client, + email: EmailSender, + url_ttl_seconds: int, + package_key_prefix: str = "bulk-downloads", + ) -> None: + self._uploaded_files = uploaded_files + self._addresses = addresses + self._documents = documents + self._packages = packages + self._email = email + self._url_ttl_seconds = url_ttl_seconds + self._package_key_prefix = package_key_prefix + + def run( + self, + *, + landlord_property_ids: Sequence[str], + 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]) + + skipped: list[SkippedDocument] = list(plan.skipped) + + if not plan.entries: + 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 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 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) + + self._email.send( + to=recipient_email, + subject="Your document download is ready", + body=( + f"Your document download is ready. It contains " + f"{included} document(s).\n\n" + f"Download it here (link valid for " + f"{self._url_ttl_seconds // 60} minutes):\n{url}" + ), + ) + return DownloadPackageResult( + presigned_url=url, + package_s3_key=key, + 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]: + if file.landlord_property_id is None: + return None + 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), + ) diff --git a/repositories/email/__init__.py b/repositories/email/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/repositories/email/email_sender.py b/repositories/email/email_sender.py new file mode 100644 index 000000000..2d6ac8725 --- /dev/null +++ b/repositories/email/email_sender.py @@ -0,0 +1,15 @@ +"""The outbound-email port (ADR-0059). + +A narrow interface the orchestrator depends on to notify a user, so the +transport (SES over SMTP today) stays swappable and testable behind it. +""" + +from __future__ import annotations + +from typing import Protocol + + +class EmailSender(Protocol): + def send(self, *, to: str, subject: str, body: str) -> None: + """Send a plain-text email to ``to``.""" + ... diff --git a/repositories/property/landlord_address_resolver.py b/repositories/property/landlord_address_resolver.py new file mode 100644 index 000000000..cf051c0bd --- /dev/null +++ b/repositories/property/landlord_address_resolver.py @@ -0,0 +1,37 @@ +"""Resolves a property's display address from its `landlord_property_id` +(ADR-0060), for naming the folders in a Download Package. + +The address lives on the FE-owned `property` table (`uploaded_files` carries no +`property_id` — a known future gap). Addresses for the 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 property row or address. +""" + +from __future__ import annotations + +from typing import Sequence + +from sqlalchemy import select +from sqlmodel import Session, col + +from infrastructure.postgres.property_table import PropertyRow + + +class LandlordAddressResolver: + _FALLBACK = "address unavailable" + + def __init__( + self, session: Session, landlord_property_ids: Sequence[str] + ) -> None: + stmt = select(PropertyRow).where( + col(PropertyRow.landlord_property_id).in_(list(landlord_property_ids)) + ) + rows = session.execute(stmt).scalars().all() # pyright: ignore[reportDeprecated] + self._by_id: dict[str, str] = { + row.landlord_property_id: row.address + for row in rows + 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) diff --git a/repositories/uploaded_file/uploaded_file_postgres_repository.py b/repositories/uploaded_file/uploaded_file_postgres_repository.py index 9bf4a3d3c..82c711231 100644 --- a/repositories/uploaded_file/uploaded_file_postgres_repository.py +++ b/repositories/uploaded_file/uploaded_file_postgres_repository.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Optional +from typing import Optional, Sequence from sqlalchemy import select from sqlmodel import Session, col @@ -24,5 +24,17 @@ class UploadedFilePostgresRepository: ) return self._session.execute(stmt).scalars().one_or_none() # pyright: ignore[reportDeprecated] + 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)) + ) + return list(self._session.execute(stmt).scalars().all()) # pyright: ignore[reportDeprecated] + def insert(self, uploaded_file: UploadedFile) -> None: self._session.add(uploaded_file) diff --git a/tests/backend/app/documents/__init__.py b/tests/backend/app/documents/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/backend/app/documents/test_bulk_download_router.py b/tests/backend/app/documents/test_bulk_download_router.py new file mode 100644 index 000000000..c21da8ca1 --- /dev/null +++ b/tests/backend/app/documents/test_bulk_download_router.py @@ -0,0 +1,240 @@ +"""Tests for the Bulk Document Download trigger endpoint (ADR-0060). + +Session, SQS and requester-email seams are dependency-injected; tests run +against the ephemeral Postgres and record message bodies instead of calling AWS. +""" + +import json +from collections.abc import Iterator +from dataclasses import dataclass, field +from typing import Any +from uuid import UUID + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import Engine +from sqlmodel import Session + +from backend.app.db.models.hubspot_deal_data import HubspotDealData +from backend.app.dependencies import validate_token +from backend.app.documents import router as documents_router +from backend.app.documents.router import ( + get_message_sender, + get_requesting_user_email, + get_session, +) +from infrastructure.postgres.property_table import PropertyRow +from infrastructure.postgres.task_table import TaskRow +from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository + +PORTFOLIO_ID = 814 +RECIPIENT = "tester@example.com" + + +@dataclass +class Api: + client: TestClient + engine: Engine + sent_bodies: list[str] = field(default_factory=list) + + def seed_task(self, selection: dict[str, Any]) -> UUID: + """Create the app-owned task with the FE-written selection config on + ``task.inputs`` (ADR-0060).""" + with Session(self.engine) as session: + row = TaskRow( + task_source="app:bulk_document_download", + inputs=json.dumps(selection), + ) + session.add(row) + session.commit() + session.refresh(row) + return row.id + + def seed_property(self, landlord_property_id: str) -> int: + with Session(self.engine) as session: + row = PropertyRow( + portfolio_id=PORTFOLIO_ID, landlord_property_id=landlord_property_id + ) + session.add(row) + session.commit() + session.refresh(row) + assert row.id is not None + return row.id + + def seed_deal(self, project_code: str, landlord_property_id: str) -> None: + """A hubspot_deal_data row carrying the project↔property link the + project-code selection resolves against (ADR-0060).""" + with Session(self.engine) as session: + session.add( + HubspotDealData( + deal_id=f"deal-{project_code}-{landlord_property_id}", + project_code=project_code, + landlord_property_id=landlord_property_id, + ) + ) + session.commit() + + def subtask_inputs(self, task_id: UUID) -> list[dict[str, Any]]: + with Session(self.engine) as session: + subtasks = SubTaskPostgresRepository(session).list_by_task(task_id) + return [st.inputs or {} for st in subtasks] + + +@pytest.fixture +def api(db_engine: Engine) -> Api: + app = FastAPI() + app.include_router(documents_router.router, prefix="/v1") + harness = Api(client=TestClient(app), engine=db_engine) + + def _session() -> Iterator[Session]: + with Session(db_engine) as session: + yield session + + app.dependency_overrides[get_session] = _session + app.dependency_overrides[get_message_sender] = lambda: harness.sent_bodies.append + app.dependency_overrides[get_requesting_user_email] = lambda: RECIPIENT + app.dependency_overrides[validate_token] = lambda: "test-token" + return harness + + +def _trigger(api: Api, task_id: UUID) -> Any: + return api.client.post( + "/v1/documents/bulk-download", json={"task_id": str(task_id)} + ) + + +def test_pins_the_recipe_and_enqueues_one_message(api: Api) -> None: + # arrange — two selected properties, pinned as the FE's task.inputs config. + p1 = api.seed_property("LP1") + p2 = api.seed_property("LP2") + task_id = api.seed_task( + {"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]} + ) + + # act + response = _trigger(api, task_id) + + # assert — accepted; one sub_task pins the recipe; one message enqueued. + assert response.status_code == 202 + inputs = api.subtask_inputs(task_id) + assert len(inputs) == 1 + assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2"] + assert inputs[0]["recipient_email"] == RECIPIENT + assert len(api.sent_bodies) == 1 + message = json.loads(api.sent_bodies[0]) + assert message["task_id"] == str(task_id) + assert "subtask_id" in message + + +def test_resolves_all_properties_for_the_given_project_codes(api: Api) -> None: + # arrange — two projects the user selects, plus one they don't; the + # project↔property grain lives on hubspot_deal_data, not `property`. + api.seed_deal("PROJ-A", "LP1") + api.seed_deal("PROJ-A", "LP2") + api.seed_deal("PROJ-B", "LP3") + api.seed_deal("PROJ-C", "LP9") # a different project — must not appear + task_id = api.seed_task({"project_codes": ["PROJ-A", "PROJ-B"]}) + + # act + response = _trigger(api, task_id) + + # assert — the union of the selected projects' landlord_property_ids is pinned. + assert response.status_code == 202 + inputs = api.subtask_inputs(task_id) + assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2", "LP3"] + + +def test_unions_project_codes_with_hand_picked_property_ids(api: Api) -> None: + # arrange — one project plus a hand-picked property from outside it. + api.seed_deal("PROJ-A", "LP1") + api.seed_deal("PROJ-A", "LP2") + hand_picked = api.seed_property("LP5") + task_id = api.seed_task( + {"project_codes": ["PROJ-A"], "property_ids": [hand_picked]} + ) + + # act + response = _trigger(api, task_id) + + # assert — both sources contribute; the set is de-duplicated and sorted. + assert response.status_code == 202 + inputs = api.subtask_inputs(task_id) + assert inputs[0]["landlord_property_ids"] == ["LP1", "LP2", "LP5"] + + +def test_deal_rows_with_no_landlord_property_id_are_dropped(api: Api) -> None: + # arrange — a project whose deal rows carry no landlord_property_id can match + # no files (files are keyed on landlord_property_id), so the selection is empty. + api.seed_deal("PROJ-A", "LP1") + with Session(api.engine) as session: + session.add(HubspotDealData(deal_id="deal-null", project_code="PROJ-A")) + session.commit() + task_id = api.seed_task({"project_codes": ["PROJ-A"]}) + + # act + response = _trigger(api, task_id) + + # assert — only the row that carries an id contributes. + assert response.status_code == 202 + inputs = api.subtask_inputs(task_id) + assert inputs[0]["landlord_property_ids"] == ["LP1"] + + +def test_rejects_a_project_code_that_resolves_to_no_properties(api: Api) -> None: + # arrange — a project with no matching deal rows at all. + task_id = api.seed_task({"project_codes": ["PROJ-EMPTY"]}) + + # act + response = _trigger(api, task_id) + + # assert — refused; nothing created or sent. + assert response.status_code == 400 + assert api.subtask_inputs(task_id) == [] + assert api.sent_bodies == [] + + +def test_rejects_a_selection_over_the_cap(api: Api, monkeypatch: Any) -> None: + # arrange — cap of 1, two selected properties. + monkeypatch.setattr(documents_router, "MAX_PROPERTIES", 1) + p1 = api.seed_property("LP1") + p2 = api.seed_property("LP2") + task_id = api.seed_task( + {"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]} + ) + + # act + response = _trigger(api, task_id) + + # assert — refused; nothing created or sent. + assert response.status_code == 400 + assert api.subtask_inputs(task_id) == [] + assert api.sent_bodies == [] + + +def test_rejects_a_task_that_already_has_a_subtask(api: Api) -> None: + # arrange — a task already triggered once. + p1 = api.seed_property("LP1") + task_id = api.seed_task({"portfolio_id": PORTFOLIO_ID, "property_ids": [p1]}) + assert _trigger(api, task_id).status_code == 202 + + # act + response = _trigger(api, task_id) + + # assert — the double-submit is refused; still one sub_task, one message. + assert response.status_code == 409 + assert len(api.subtask_inputs(task_id)) == 1 + assert len(api.sent_bodies) == 1 + + +def test_rejects_a_task_with_no_selection_config(api: Api) -> None: + # arrange — a task whose inputs carry no portfolio_id/selection. + api.seed_property("LP1") + task_id = api.seed_task({}) + + # act + response = _trigger(api, task_id) + + # assert + assert response.status_code == 400 + assert api.sent_bodies == [] diff --git a/tests/domain/bulk_document_download/__init__.py b/tests/domain/bulk_document_download/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/domain/bulk_document_download/test_package_plan.py b/tests/domain/bulk_document_download/test_package_plan.py new file mode 100644 index 000000000..4bcddc83c --- /dev/null +++ b/tests/domain/bulk_document_download/test_package_plan.py @@ -0,0 +1,94 @@ +"""The Download Package plan (ADR-0060) — pure layout of the ZIP from resolved +documents: one folder per property, latest file of each Document Type, +null-Document-Type rows skipped and reported.""" + +from datetime import datetime, timezone +from typing import Optional + +from domain.bulk_document_download.package_plan import ( + ResolvedDocument, + plan_download_package, +) + + +def _doc( + *, + landlord_property_id: str = "LP1", + address: str = "12 Oak Street", + document_type: Optional[str] = "site_note", + s3_key: str = "a.pdf", + uploaded_at: datetime = datetime(2026, 1, 1, tzinfo=timezone.utc), +) -> ResolvedDocument: + return ResolvedDocument( + landlord_property_id=landlord_property_id, + address=address, + document_type=document_type, + s3_bucket="src-bucket", + s3_key=s3_key, + uploaded_at=uploaded_at, + ) + + +def test_keeps_only_the_latest_file_of_a_document_type_in_the_property_folder() -> None: + # Arrange — two site_note files for the same property, uploaded at + # different times; only the newest should be packaged. + older = _doc( + s3_key="old.pdf", uploaded_at=datetime(2026, 1, 1, tzinfo=timezone.utc) + ) + newer = _doc( + s3_key="new.pdf", uploaded_at=datetime(2026, 6, 1, tzinfo=timezone.utc) + ) + + # Act + plan = plan_download_package([older, newer]) + + # Assert — one entry, the newer file, foldered under the property. + assert len(plan.entries) == 1 + entry = plan.entries[0] + assert entry.s3_key == "new.pdf" + assert entry.zip_path.startswith("12 Oak Street (LP1)/") + + +def test_skips_and_reports_a_document_with_no_document_type() -> None: + # Arrange — one packable file and one row whose Document Type is null + # (file_type is nullable on uploaded_files). + packable = _doc(document_type="site_note", s3_key="note.pdf") + untyped = _doc(document_type=None, s3_key="mystery.bin") + + # Act + plan = plan_download_package([packable, untyped]) + + # Assert — the untyped row is left out of the archive and reported instead. + assert [e.s3_key for e in plan.entries] == ["note.pdf"] + assert len(plan.skipped) == 1 + skipped = plan.skipped[0] + assert skipped.s3_key == "mystery.bin" + assert skipped.landlord_property_id == "LP1" + + +def test_packaged_filename_is_the_document_type_with_the_original_extension() -> None: + # Arrange — a photo pack stored under a prefixed key with a .zip extension; + # the file inside the archive should be named by type but stay openable. + doc = _doc(document_type="photo_pack", s3_key="surveys/2026/pack.zip") + + # Act + plan = plan_download_package([doc]) + + # Assert + assert plan.entries[0].zip_path == "12 Oak Street (LP1)/photo_pack.zip" + + +def test_latest_per_type_is_scoped_per_property_each_gets_its_own_folder() -> None: + # Arrange — two DIFFERENT properties, each with a site_note. Latest-per-type + # must not dedupe across properties: both survive, in their own folders. + first = _doc(landlord_property_id="LP1", address="12 Oak Street", s3_key="a.pdf") + second = _doc(landlord_property_id="LP2", address="9 Elm Road", s3_key="b.pdf") + + # Act + plan = plan_download_package([first, second]) + + # Assert + assert {e.zip_path for e in plan.entries} == { + "12 Oak Street (LP1)/site_note.pdf", + "9 Elm Road (LP2)/site_note.pdf", + } diff --git a/tests/infrastructure/email/__init__.py b/tests/infrastructure/email/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/infrastructure/email/test_ses_smtp_email_sender.py b/tests/infrastructure/email/test_ses_smtp_email_sender.py new file mode 100644 index 000000000..036ac239f --- /dev/null +++ b/tests/infrastructure/email/test_ses_smtp_email_sender.py @@ -0,0 +1,104 @@ +"""The SES-over-SMTP email sender (ADR-0059) — delivers a plain-text message to +the requesting user via an injected SMTP transport.""" + +from __future__ import annotations + +from email.message import EmailMessage +from types import TracebackType +from typing import Any, Optional, Type + +from infrastructure.email.ses_smtp_email_sender import SesSmtpEmailSender + + +class _FakeSmtp: + """Records what the sender hands the transport, standing in for a live SES + SMTP connection.""" + + def __init__(self, host: str, port: int) -> None: + self.host = host + self.port = port + self.messages: list[EmailMessage] = [] + self.started_tls = False + self.login_args: Optional[tuple[str, str]] = None + + def __enter__(self) -> "_FakeSmtp": + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + tb: Optional[TracebackType], + ) -> Optional[bool]: + return None + + def starttls(self) -> Any: + self.started_tls = True + + def login(self, user: str, password: str) -> Any: + self.login_args = (user, password) + + def send_message(self, message: EmailMessage) -> Any: + self.messages.append(message) + + +def test_send_delivers_a_plaintext_message_with_the_right_envelope() -> 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="Your document download is ready", + body="Download it here: https://signed-url", + ) + + # Assert — one message, correctly addressed, body carried through. + assert len(created) == 1 + assert len(created[0].messages) == 1 + message = created[0].messages[0] + assert message["To"] == "user@example.com" + assert message["From"] == "noreply@domna.homes" + assert message["Subject"] == "Your document download is ready" + assert "https://signed-url" in message.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. + 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="b") + + # Assert + assert created[0].started_tls is True + assert created[0].login_args == ("AKIA_SMTP_USER", "smtp-password") diff --git a/tests/infrastructure/test_s3_client.py b/tests/infrastructure/test_s3_client.py index bdac6be1e..13e6c4714 100644 --- a/tests/infrastructure/test_s3_client.py +++ b/tests/infrastructure/test_s3_client.py @@ -34,3 +34,33 @@ def test_get_object_returns_bytes_written_by_put_object(s3_client: S3Client) -> def test_bucket_property_exposes_configured_bucket(s3_client: S3Client) -> None: # act / assert assert s3_client.bucket == BUCKET + + +def test_generate_presigned_url_signs_a_get_for_the_object(s3_client: S3Client) -> None: + # arrange + s3_client.put_object("bulk-downloads/pkg.zip", b"zip-bytes") + + # act + url = s3_client.generate_presigned_url("bulk-downloads/pkg.zip", expires_in=3600) + + # assert — a signed GET URL for this bucket + key + assert BUCKET in url + assert "bulk-downloads/pkg.zip" in url + assert "Signature" in url + + +def test_upload_file_streams_a_local_file_to_s3( + s3_client: S3Client, tmp_path: object +) -> None: + # arrange — a file on local disk (stands in for the /tmp ZIP) + from pathlib import Path + + local = Path(str(tmp_path)) / "package.zip" + local.write_bytes(b"ZIP-CONTENT") + + # act + uri = s3_client.upload_file(str(local), "bulk-downloads/package.zip") + + # assert — the object landed with the file's bytes + assert uri == f"s3://{BUCKET}/bulk-downloads/package.zip" + assert s3_client.get_object("bulk-downloads/package.zip") == b"ZIP-CONTENT" 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 new file mode 100644 index 000000000..d9b3a0d23 --- /dev/null +++ b/tests/orchestration/test_bulk_document_download_orchestrator.py @@ -0,0 +1,244 @@ +"""The Bulk Document Download orchestrator (ADR-0060): gather a property set's +files, lay them out, zip + upload, presign, and email the requester.""" + +from __future__ import annotations + +import io +import zipfile +from collections.abc import Iterator +from datetime import datetime, timezone + +import pytest +from moto import mock_aws + +from domain.tasks.subtasks import SubTaskFailure +from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile +from infrastructure.s3.s3_client import S3Client +from orchestration.bulk_document_download_orchestrator import ( + BulkDocumentDownloadOrchestrator, +) +from tests.infrastructure import make_boto_client + +DATA_BUCKET = "data-bucket" + + +class _FakeUploadedFiles: + def __init__(self, files: list[UploadedFile]) -> None: + self._files = files + + def by_landlord_property_ids( + self, landlord_property_ids: object + ) -> list[UploadedFile]: + wanted = set(landlord_property_ids) # type: ignore[call-overload] + return [f for f in self._files if f.landlord_property_id in wanted] + + +class _FakeAddresses: + def __init__(self, mapping: dict[str, str]) -> None: + self._mapping = mapping + + def address_for(self, landlord_property_id: str) -> str: + return self._mapping[landlord_property_id] + + +class _FakeDocuments: + def __init__(self, blobs: dict[tuple[str, str], bytes]) -> None: + self._blobs = blobs + + 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: + def __init__(self) -> None: + self.sent: list[tuple[str, str, str]] = [] + + def send(self, *, to: str, subject: str, body: str) -> None: + self.sent.append((to, subject, body)) + + +def _uploaded_file( + 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), + landlord_property_id=landlord_property_id, + file_type=file_type.value, + ) + + +@pytest.fixture +def packages() -> Iterator[S3Client]: + with mock_aws(): + boto_client = make_boto_client("s3") + boto_client.create_bucket(Bucket=DATA_BUCKET) + yield S3Client(boto_client, DATA_BUCKET) + + +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")] + email = _RecordingEmail() + 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=email, + url_ttl_seconds=3600, + ) + + # Act + result = orchestrator.run( + landlord_property_ids=["LP1"], + recipient_email="user@example.com", + package_name="portfolio-42", + ) + + # Assert — one file packaged, at the planned path with the source bytes. + assert result.included == 1 + archive = packages.get_object(result.package_s3_key) + with zipfile.ZipFile(io.BytesIO(archive)) as zf: + assert zf.read("12 Oak Street (LP1)/site_note.pdf") == b"PDF-BYTES" + + # ...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] + assert to == "user@example.com" + assert result.presigned_url in body + + +def test_a_selection_with_no_documents_is_a_recorded_failure( + packages: S3Client, +) -> None: + # Arrange — the selection resolves to no packageable documents. + email = _RecordingEmail() + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=_FakeUploadedFiles([]), + addresses=_FakeAddresses({}), + documents=_FakeDocuments({}), + packages=packages, + email=email, + url_ttl_seconds=3600, + ) + + # Act / Assert — a recorded SubTask failure, and nobody is emailed an empty + # package. + with pytest.raises(SubTaskFailure): + orchestrator.run( + landlord_property_ids=["LP1"], + recipient_email="user@example.com", + package_name="empty", + ) + assert email.sent == [] + + +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, + ) + email = _RecordingEmail() + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=_FakeUploadedFiles([good, untyped]), + 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="mixed", + ) + + # Assert — the good file is packaged and delivered; the untyped one is + # reported, not fatal. + assert result.included == 1 + assert [s.s3_key for s in result.skipped] == ["mystery.bin"] + 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_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"] diff --git a/tests/repositories/property/test_landlord_address_resolver.py b/tests/repositories/property/test_landlord_address_resolver.py new file mode 100644 index 000000000..7dcbb98da --- /dev/null +++ b/tests/repositories/property/test_landlord_address_resolver.py @@ -0,0 +1,35 @@ +from collections.abc import Iterator + +import pytest +from sqlalchemy import Engine +from sqlmodel import Session + +from infrastructure.postgres.property_table import PropertyRow +from repositories.property.landlord_address_resolver import LandlordAddressResolver + + +@pytest.fixture +def session(db_engine: Engine) -> Iterator[Session]: + with Session(db_engine) as s: + yield s + + +def test_resolves_known_addresses_and_falls_back_for_unknowns( + session: Session, +) -> None: + # arrange — two properties with addresses; a third id has no property row. + session.add( + PropertyRow(portfolio_id=1, landlord_property_id="LP1", address="12 Oak Street") + ) + session.add( + PropertyRow(portfolio_id=1, landlord_property_id="LP2", address="9 Elm Road") + ) + session.flush() + + # act + resolver = LandlordAddressResolver(session, ["LP1", "LP2", "LP3"]) + + # assert + assert resolver.address_for("LP1") == "12 Oak Street" + assert resolver.address_for("LP2") == "9 Elm Road" + assert resolver.address_for("LP3") == "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 5300c0209..40fc1773f 100644 --- a/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py +++ b/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py @@ -78,3 +78,38 @@ def test_does_not_return_row_with_different_file_type(db_engine: Engine) -> None # Assert assert result is None + + +def _landlord_file( + landlord_property_id: str, + s3_file_key: str, + file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE, +) -> UploadedFile: + 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, + file_type=file_type.value, + ) + + +def test_by_landlord_property_ids_returns_only_requested_properties( + db_engine: Engine, +) -> None: + # Arrange — files for three properties; only 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.commit() + + # Act + with Session(db_engine) as session: + found = UploadedFilePostgresRepository(session).by_landlord_property_ids( + ["LP1", "LP2"] + ) + result = {(f.landlord_property_id, f.s3_file_key) for f in found} + + # Assert + assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")}