FastAPI trigger route: POST /v1/documents/bulk-download 🟩

Resolves the authenticated requester's email (ADR-0059), resolves + caps the
property selection, pins the recipe (landlord_property_ids, recipient_email,
package_name) onto one pre-created sub_task (raw SQL, dodging the mirror
double-registration), and enqueues one message to the worker. Refuses
double-submits (409) and oversized selections (400).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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"

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()

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"}

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

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

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