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..1ef26eb04 --- /dev/null +++ b/backend/app/documents/download_tasks.py @@ -0,0 +1,91 @@ +"""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, Optional +from uuid import UUID + +from sqlalchemy import text +from sqlmodel import Session + + +class DocumentDownloadTasks: + def __init__(self, session: Session) -> None: + self._session = session + + 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, + portfolio_id: int, + property_ids: Optional[list[int]], + select_all: bool, + ) -> tuple[int, list[str]]: + """Resolve the selection to (selected property count, the distinct + ``landlord_property_id`` values that carry a downloadable identifier). + + The count is the cap basis (property count, ADR-0060); files are matched + by ``landlord_property_id`` (the missing ``property_id`` on + ``uploaded_files`` is a known future gap), so a selected property with a + null ``landlord_property_id`` contributes no files. + """ + if select_all: + rows = self._session.connection().execute( + text( + "SELECT id, landlord_property_id FROM property" + " WHERE portfolio_id = :portfolio_id" + ), + {"portfolio_id": portfolio_id}, + ) + else: + rows = self._session.connection().execute( + text( + "SELECT id, landlord_property_id FROM property" + " WHERE portfolio_id = :portfolio_id AND id = ANY(:ids)" + ), + {"portfolio_id": portfolio_id, "ids": property_ids or []}, + ) + resolved = rows.all() + landlord_property_ids = sorted( + {row[1] for row in resolved if row[1] is not None} + ) + return len(resolved), landlord_property_ids + + def create_download_subtask( + self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any] + ) -> None: + """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).""" + now = datetime.now(timezone.utc) + self._session.connection().execute( + text( + "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)" + " VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)" + ), + { + "id": subtask_id, + "task_id": task_id, + "inputs": json.dumps(recipe), + "updated_at": now, + }, + ) + self._session.commit() diff --git a/backend/app/documents/router.py b/backend/app/documents/router.py new file mode 100644 index 000000000..7d3eea4db --- /dev/null +++ b/backend/app/documents/router.py @@ -0,0 +1,132 @@ +"""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 = getattr(user, "email", None) + if email is None and isinstance(user, dict): + email = user.get("email") + 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." + ), + ) + + if not body.select_all and not body.property_ids: + raise HTTPException( + status_code=400, + detail="Provide property_ids or set select_all.", + ) + + property_count, landlord_property_ids = tasks.resolve_selection( + body.portfolio_id, body.property_ids, body.select_all + ) + if property_count == 0: + raise HTTPException( + status_code=400, + detail=( + f"The selection resolves to no properties in portfolio " + f"{body.portfolio_id}." + ), + ) + if property_count > MAX_PROPERTIES: + raise HTTPException( + status_code=400, + detail=( + f"The selection has {property_count} properties, over the " + f"{MAX_PROPERTIES} limit — narrow your selection." + ), + ) + if not landlord_property_ids: + raise HTTPException( + status_code=400, + detail=( + "None of the selected properties have a landlord_property_id, " + "so no documents can be matched." + ), + ) + + subtask_id = uuid4() + recipe = { + "landlord_property_ids": landlord_property_ids, + "recipient_email": recipient_email, + "package_name": f"portfolio-{body.portfolio_id}-{body.task_id}", + } + tasks.create_download_subtask(body.task_id, subtask_id, recipe) + + 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..4d9a06247 --- /dev/null +++ b/backend/app/documents/schemas.py @@ -0,0 +1,19 @@ +from typing import Optional +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 (its own request lands in + ``task.inputs``) and passes ``task_id`` here. Either an explicit + ``property_ids`` selection or ``select_all`` (the whole portfolio) must be + given. + """ + + task_id: UUID + portfolio_id: int + property_ids: Optional[list[int]] = None + select_all: bool = False 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/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..b07d6baf3 --- /dev/null +++ b/tests/backend/app/documents/test_bulk_download_router.py @@ -0,0 +1,140 @@ +"""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 + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import Engine +from sqlmodel import Session + +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 domain.tasks.tasks import Task +from infrastructure.postgres.property_table import PropertyRow +from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository +from repositories.tasks.task_postgres_repository import TaskPostgresRepository + +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) -> Task: + with Session(self.engine) as session: + return TaskPostgresRepository(session).create( + Task.create(task_source="app:bulk_document_download") + ) + + 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 subtask_inputs(self, task: Task) -> 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: Task, property_ids: list[int]) -> Any: + return api.client.post( + "/v1/documents/bulk-download", + json={ + "task_id": str(task.id), + "portfolio_id": PORTFOLIO_ID, + "property_ids": property_ids, + }, + ) + + +def test_pins_the_recipe_and_enqueues_one_message(api: Api) -> None: + # arrange — two selected properties, matched by landlord_property_id. + task = api.seed_task() + p1 = api.seed_property("LP1") + p2 = api.seed_property("LP2") + + # act + response = _trigger(api, task, [p1, p2]) + + # assert — accepted; one sub_task pins the recipe; one message enqueued. + assert response.status_code == 202 + inputs = api.subtask_inputs(task) + 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_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) + task = api.seed_task() + p1 = api.seed_property("LP1") + p2 = api.seed_property("LP2") + + # act + response = _trigger(api, task, [p1, p2]) + + # assert — refused; nothing created or sent. + assert response.status_code == 400 + assert api.subtask_inputs(task) == [] + assert api.sent_bodies == [] + + +def test_rejects_a_task_that_already_has_a_subtask(api: Api) -> None: + # arrange — a task already triggered once. + task = api.seed_task() + p1 = api.seed_property("LP1") + assert _trigger(api, task, [p1]).status_code == 202 + + # act + response = _trigger(api, task, [p1]) + + # assert — the double-submit is refused; still one sub_task, one message. + assert response.status_code == 409 + assert len(api.subtask_inputs(task)) == 1 + assert len(api.sent_bodies) == 1