diff --git a/backend/app/documents/download_tasks.py b/backend/app/documents/download_tasks.py index 1ef26eb04..9c4940407 100644 --- a/backend/app/documents/download_tasks.py +++ b/backend/app/documents/download_tasks.py @@ -8,7 +8,7 @@ Modelling Run distributor works around — see ``modelling/run_tasks.py``). import json from datetime import datetime, timezone -from typing import Any, Optional +from typing import Any, Optional, cast from uuid import UUID from sqlalchemy import text @@ -19,6 +19,23 @@ 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.""" diff --git a/backend/app/documents/router.py b/backend/app/documents/router.py index f35974296..06bc24723 100644 --- a/backend/app/documents/router.py +++ b/backend/app/documents/router.py @@ -84,21 +84,30 @@ async def bulk_download( ), ) - if not body.select_all and not body.property_ids: + config = tasks.read_selection_config(body.task_id) + portfolio_id = config.get("portfolio_id") + property_ids = config.get("property_ids") + select_all = bool(config.get("select_all", False)) + if not isinstance(portfolio_id, int): raise HTTPException( status_code=400, - detail="Provide property_ids or set select_all.", + detail="task.inputs is missing an integer portfolio_id.", + ) + if not select_all and not property_ids: + raise HTTPException( + status_code=400, + detail="task.inputs must provide property_ids or set select_all.", ) property_count, landlord_property_ids = tasks.resolve_selection( - body.portfolio_id, body.property_ids, body.select_all + portfolio_id, property_ids, 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}." + f"{portfolio_id}." ), ) if property_count > MAX_PROPERTIES: @@ -122,7 +131,7 @@ async def bulk_download( recipe = { "landlord_property_ids": landlord_property_ids, "recipient_email": recipient_email, - "package_name": f"portfolio-{body.portfolio_id}-{body.task_id}", + "package_name": f"portfolio-{portfolio_id}-{body.task_id}", } tasks.create_download_subtask(body.task_id, subtask_id, recipe) diff --git a/backend/app/documents/schemas.py b/backend/app/documents/schemas.py index 4d9a06247..0d60006d8 100644 --- a/backend/app/documents/schemas.py +++ b/backend/app/documents/schemas.py @@ -1,4 +1,3 @@ -from typing import Optional from uuid import UUID from pydantic import BaseModel @@ -7,13 +6,11 @@ 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. + The front end creates the app-owned Task first and writes the **selection + config** into ``task.inputs`` (a JSON object of + ``{portfolio_id, property_ids?, select_all?}`` — so a large hand-picked set + never travels in an HTTP body), then passes only the ``task_id`` here. The + backend reads the selection from ``task.inputs``. """ task_id: UUID - portfolio_id: int - property_ids: Optional[list[int]] = None - select_all: bool = False diff --git a/docs/adr/0060-bulk-document-download-package-model.md b/docs/adr/0060-bulk-document-download-package-model.md index a220033ff..103a01b47 100644 --- a/docs/adr/0060-bulk-document-download-package-model.md +++ b/docs/adr/0060-bulk-document-download-package-model.md @@ -22,13 +22,20 @@ 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 FastAPI route creates the `tasks` - row, resolves the selection to a concrete `property_id` set (expanding "all - properties" of a portfolio), **pins that set plus the recipient email into - `tasks.inputs`** (TEXT/JSON — a large id list is why this lives in `inputs`, - not the 256 KB SQS body), pre-creates **one** `sub_task`, and drops one SQS - message (`task_id`, `sub_task_id`). The new `applications/bulk_document_download` - Lambda runs in **attach mode**; `TaskOrchestrator` owns status + roll-up. +- **Trigger & lifecycle (ADR-0055).** The **front end** creates the app-owned + `tasks` row and writes the **selection config** — + `{portfolio_id, property_ids?, select_all?}` — into `tasks.inputs` (TEXT/JSON, + FE-owned), then calls the FastAPI route with **only the `task_id`**. A large + hand-picked selection therefore never travels in an HTTP body. The route + **reads the selection from `tasks.inputs`**, resolves it to the distinct + `landlord_property_id` set, caps it, 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 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/tests/backend/app/documents/test_bulk_download_router.py b/tests/backend/app/documents/test_bulk_download_router.py index b07d6baf3..ae5f38725 100644 --- a/tests/backend/app/documents/test_bulk_download_router.py +++ b/tests/backend/app/documents/test_bulk_download_router.py @@ -8,6 +8,7 @@ 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 @@ -22,10 +23,9 @@ from backend.app.documents.router import ( get_requesting_user_email, get_session, ) -from domain.tasks.tasks import Task from infrastructure.postgres.property_table import PropertyRow +from infrastructure.postgres.task_table import TaskRow from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository -from repositories.tasks.task_postgres_repository import TaskPostgresRepository PORTFOLIO_ID = 814 RECIPIENT = "tester@example.com" @@ -37,11 +37,18 @@ class Api: engine: Engine sent_bodies: list[str] = field(default_factory=list) - def seed_task(self) -> Task: + 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: - return TaskPostgresRepository(session).create( - Task.create(task_source="app:bulk_document_download") + 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: @@ -54,9 +61,9 @@ class Api: assert row.id is not None return row.id - def subtask_inputs(self, task: Task) -> list[dict[str, Any]]: + 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) + subtasks = SubTaskPostgresRepository(session).list_by_task(task_id) return [st.inputs or {} for st in subtasks] @@ -77,64 +84,76 @@ def api(db_engine: Engine) -> Api: return harness -def _trigger(api: Api, task: Task, property_ids: list[int]) -> Any: +def _trigger(api: Api, task_id: UUID) -> Any: return api.client.post( - "/v1/documents/bulk-download", - json={ - "task_id": str(task.id), - "portfolio_id": PORTFOLIO_ID, - "property_ids": property_ids, - }, + "/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, matched by landlord_property_id. - task = api.seed_task() + # 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, [p1, p2]) + 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) + 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 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") + task_id = api.seed_task( + {"portfolio_id": PORTFOLIO_ID, "property_ids": [p1, p2]} + ) # act - response = _trigger(api, task, [p1, p2]) + response = _trigger(api, task_id) # assert — refused; nothing created or sent. assert response.status_code == 400 - assert api.subtask_inputs(task) == [] + 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. - task = api.seed_task() p1 = api.seed_property("LP1") - assert _trigger(api, task, [p1]).status_code == 202 + 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, [p1]) + 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)) == 1 + 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 == []