Model/backlog/ventilation-audit-generator.md

PRD: Ventilation Audit Generator from MagicPlan

Problem Statement

When a surveyor completes a MagicPlan survey for a property, the resulting floor plan data (rooms, windows, doors, ventilation measurements) needs to be transformed into a structured ventilation audit spreadsheet. Currently this transformation is manual — someone must extract plan data and populate a report by hand, which is slow and error-prone.

Solution

An AWS Lambda (audit-generator) triggered via SQS receives a HubSpot deal ID, fetches the parsed MagicPlan Plan from the database, populates a pre-formatted .xlsx template with plan data, uploads the result to S3, and records it in uploaded_files. The populated spreadsheet is then accessible to the UI so the user knows an audit file exists for that deal.

User Stories

1. As a coordinator, I want clicking a button in the UI to trigger generation of a ventilation audit spreadsheet, so that I do not have to manually populate it from the floor plan.
2. As a coordinator, I want the audit spreadsheet to be automatically populated with room, window, and door data from the MagicPlan survey, so that the data entry step is eliminated.
3. As a coordinator, I want the system to use a pre-formatted .xlsx template when generating the audit, so that conditional formatting and layout are preserved without requiring code changes.
4. As a coordinator, I want the UI to indicate whether a ventilation audit already exists for a deal, so that I avoid triggering duplicate generation unnecessarily.
5. As a coordinator, I want re-triggering generation to overwrite the previous audit file, so that I can regenerate after a corrected survey is uploaded.
6. As an engineer, I want the lambda to raise a clear error if no MagicPlan JSON has been uploaded for the deal, so that misconfigured triggers are diagnosed quickly.
7. As an engineer, I want the lambda to raise a distinct error if a MagicPlan JSON exists but has not yet been parsed into the database, so that timing issues are distinguishable from missing data.
8. As an engineer, I want the generated spreadsheet recorded in uploaded_files with a VENTILATION_AUDIT file type, so that the UI and other systems can query for its existence.
9. As an engineer, I want the lambda to follow the @subtask_handler() pattern, so that it integrates with the task orchestration system and benefits from standard error handling and observability.

Implementation Decisions

• Lambda pattern: @subtask_handler() decorator. Trigger body contains task_id, sub_task_id, and hubspot_deal_id.

• MAGIC_PLAN_JSON lookup: Query uploaded_files filtered by hubspot_deal_id and file_type = MAGIC_PLAN_JSON, ordered by s3_upload_timestamp DESC, taking the most recent row. Rationale: a re-upload supersedes the earlier file.

• Plan retrieval: Use the existing MagicPlanPostgresRepository.get_plan_by_uploaded_file_id to fetch the parsed domain Plan from postgres. The lambda does not re-parse from S3 — that is the magic_plan lambda's responsibility.

• Error handling — two distinct cases:

  • No uploaded_files row found → raise with message indicating no MagicPlan has been uploaded for this deal.
  • Row found but get_plan_by_uploaded_file_id returns None → raise with message indicating the plan has been uploaded but not yet parsed.
  • Both use the same exception type; distinct messages enable diagnosis in CloudWatch.

• Spreadsheet generation:

  • Format: .xlsx via openpyxl.
  • The template d1_ventilation_template.xlsx is bundled with the lambda at applications/audit-generator/d1_ventilation_template.xlsx and loaded from the deployment package via importlib.resources or a path relative to the handler file. No S3 round-trip for the template.
  • The template is loaded with openpyxl.load_workbook(path) (default data_only=False to preserve formulas), populated, and serialised to bytes via BytesIO for upload.
  • Cell targeting uses fixed column letters (see Spreadsheet Layout below). Named ranges are not defined in the template.
  • The template has formulas in columns J (=H*I), N (=J*M), S (=Q*R), and Y (=W*X) — the lambda does not write to these cells; they are calculated by Excel/Sheets when the file is opened.
  • The template has 50 data rows (rows 6–55), extended programmatically. The footer merge sits at A56:Z56; legend rows at 57–60.

• Output S3 key: documents/hubspot_deal_id/{hubspot_deal_id}/ventilation_audit.xlsx. Re-running the lambda overwrites the previous file.

• Operation order: S3 upload first, then uploaded_files DB insert. An orphaned S3 file on DB failure is harmless and will be overwritten on retry. A DB record pointing to a non-existent file is worse.

• New enum values (added to FileTypeEnum and FileSourceEnum):

  • FileTypeEnum.VENTILATION_AUDIT = "ventilation_audit"
  • FileSourceEnum.AUDIT_GENERATOR = "audit_generator"

• DDD migration of UploadedFile: The existing backend/app/db/models/uploaded_file.py (SQLAlchemy Base) is replaced by infrastructure/postgres/uploaded_file_table.py (SQLModel). FileTypeEnum, FileSourceEnum, and UploadedFile all move there. The class name UploadedFile is kept (no Model suffix — there is no domain counterpart). All seven consumers update their import path; backend/app/db/models/uploaded_file.py is deleted. Because UploadedFile is now registered on SQLModel.metadata, the shared tests/conftest.py db_engine fixture must emit CREATE TYPE IF NOT EXISTS for file_type and file_source via raw SQL before calling SQLModel.metadata.create_all(engine) — otherwise the table creation fails for all integration tests. The dedicated per-test conftest approach (Question 6) is therefore superseded.

• New UploadedFileRepository: A new repository (UploadedFilePostgresRepository) is introduced with a get_latest_by_hubspot_deal_id(hubspot_deal_id: str, file_type: FileTypeEnum) -> Optional[UploadedFile] method. Queries uploaded_files filtered by hubspot_deal_id and file_type, ordered by s3_upload_timestamp DESC, returning the most recent row.

• Session management: A dedicated AuditGeneratorUnitOfWork context manager (standalone — does not inherit from PostgresUnitOfWork or UnitOfWork) holds uploaded_file: UploadedFilePostgresRepository and magic_plan: MagicPlanPostgresRepository, both bound to the same session. Opens the session on __enter__, rolls back and closes on __exit__, exposes commit(). The handler holds a module-scoped engine (reused across warm Lambda invocations) and passes a session_factory callable to AuditGeneratorUnitOfWork — the session is created fresh per invocation and never long-lived.

• Idempotency: No duplicate guard. uploaded_files is append-only — the lambda always inserts a new row; rows are never updated or deleted. The S3 file is always overwritten at the fixed key. The UI and any future queries treat the most recent row by s3_upload_timestamp as authoritative.

• Environment variables:

  • S3_BUCKET_NAME (shared convention)
  • DATABASE_URL (shared convention)

• Trigger: The SQS message is sent by a UI action in a separate repo. No SQS publishing client is required in this PR.

Testing Decisions

Good tests assert observable outputs given controlled inputs — they do not assert on internal call sequences or implementation details. Prefer mocking at the boundary of the system under test, not inside it.

Handler tests (tests/applications/audit_generator/test_audit_generator_handler.py):

• Test that an invalid trigger body raises ValidationError.
• Test that the orchestrator is constructed with values derived from env vars and the trigger body.
• Test that the handler returns the expected value on success.
• Use handler.__wrapped__ to bypass the @subtask_handler decorator (prior art: test_magic_plan_handler.py).

Orchestrator tests (tests/orchestration/audit_generator/test_audit_generator_orchestrator.py):

• Mock S3Client with MagicMock(spec=S3Client). Mock the AuditGeneratorUnitOfWork factory: the factory returns a mock UoW whose __enter__ returns itself and whose .uploaded_file and .magic_plan attributes are mock repos.
• Test happy path: correct S3 key used for output upload; uploaded_files insert called with correct file_type and file_source; uow.commit() called.
• Test error path: raises with appropriate message when uploaded_file_repo.get_latest_by_hubspot_deal_id returns None.
• Test error path: raises with appropriate message when magic_plan_repo.get_plan_by_uploaded_file_id returns None.

Repository tests (tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py):

• Integration tests using the shared db_engine fixture. The fixture already calls SQLModel.metadata.create_all(engine); after the DDD migration UploadedFile is in SQLModel.metadata, so no dedicated conftest is needed. The shared tests/conftest.py must emit CREATE TYPE IF NOT EXISTS for file_type and file_source before create_all.
• Test that get_latest_by_hubspot_deal_id returns the most recent row by s3_upload_timestamp when multiple rows with the same file_type exist.
• Test that it returns None when no matching row exists.
• Test that it filters correctly by file_type (a row with a different file_type is not returned).

Out of Scope

• The SQS trigger — the UI button that sends the SQS message lives in a separate repo.
• Any ventilation calculation or compliance logic — the spreadsheet is populated with raw plan data only.

Spreadsheet Layout

Sheet name: D1 Ventilation. Data starts at row 6. The three series run in parallel columns — each row may contain room data, window data, and door data independently; the longest series determines the last row used.

Column	Content	Source
B	Room name	Room.name
D	Room area (m²)	Room.area_m2
G	Window location (room name)	Room.name (parent room)
H	Window width (m)	Window.width_m
I	Window height (m)	Window.height_m
J	Window area (m²)	formula =H*I — do not write
K	Opening type	WindowVentilation.opening_type
L	Number of openings	WindowVentilation.num_openings
M	% of window (decimal)	WindowVentilation.pct_openable / 100
N	Total opening area (m²)	formula =J*M — do not write
O	Blocked	leave blank (visual check by auditor)
P	Pictured	leave blank (visual check by auditor)
Q	Trickle vent effective area per vent (mm²)	WindowVentilation.trickle_vent_area_mm2
R	Number of trickle vents	WindowVentilation.num_trickle_vents
S	Total trickle vent area (mm²)	formula =Q*R — do not write
V	Door location (room name)	Room.name (parent room)
W	Door width (mm)	Door.width_mm
X	Door undercut (mm)	DoorVentilation.undercut_mm
Y	Door area (mm²)	formula =W*X — do not write

Internal doors appear once per room they connect (typically twice). WindowVentilation and DoorVentilation fields are Optional; write 0 when None so formula cells (J, N, S, Y) do not produce #VALUE! errors.

Further Notes

• The audit-generator application scaffold already exists at applications/audit-generator/ with empty handler.py and audit_generator_trigger_request.py files.
• The MagicPlanPostgresRepository.get_plan_by_uploaded_file_id method is the correct entry point for fetching the parsed plan — no S3 re-parsing is needed.
• The openpyxl library must be added to applications/audit-generator/handler/requirements.txt.
• The template (d1_ventilation_template.xlsx) has 50 data rows (rows 6–55) with formulas in columns J, N, S, Y. If a property exceeds 50 windows, rooms, or doors the lambda should raise a clear error rather than silently truncating.