Merge pull request #1060 from Hestia-Homes/feature/magicplan-trigger

MagicPlan lambda and trigger

.github/workflows/deploy_terraform.yml

@@ -537,11 +537,49 @@ jobs:
       AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
       AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
 
+  # ============================================================
+  # Build MagicPlan Lambda image
+  # ============================================================
+  magic_plan_image:
+    needs: [determine_stage, shared_terraform]
+    uses: ./.github/workflows/_build_image.yml
+    with:
+      ecr_repo: magic-plan-${{ needs.determine_stage.outputs.stage }}
+      dockerfile_path: backend/magic_plan/handler/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 MagicPlan Lambda
+  # ============================================================
+  magic_plan_lambda:
+    needs: [magic_plan_image, determine_stage]
+    uses: ./.github/workflows/_deploy_lambda.yml
+    with:
+      lambda_name: magic_plan
+      lambda_path: infrastructure/terraform/lambda/magic_plan
+      stage: ${{ needs.determine_stage.outputs.stage }}
+      ecr_repo: magic-plan-${{ needs.determine_stage.outputs.stage }}
+      image_digest: ${{ needs.magic_plan_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 }}
+      TF_VAR_magicplan_customer_id: ${{ secrets.MAGICPLAN_CUSTOMER_ID }}
+      TF_VAR_magicplan_api_key: ${{ secrets.MAGICPLAN_API_KEY }}
+
   # ============================================================
   # Deploy Hubspot ETL Lambda
   # ============================================================
   hubspot_etl_lambda:
-    needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda]
+    needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda]
     uses: ./.github/workflows/_deploy_lambda.yml
     with:
       lambda_name: hubspot-etl-to-ara

backend/app/config.py

@@ -39,6 +39,7 @@ class Settings(BaseSettings):
     ENGINE_SQS_URL: str = "changeme"
     CATEGORISATION_SQS_URL: str = "changeme"
     PASHUB_TO_ARA_SQS_URL: str = "changeme"
+    MAGICPLAN_SQS_URL: str = "changeme"
     POSTCODE_SPLITTER_SQS_URL: str = "changeme"
     COMBINER_SQS_URL: str = "changeme"
 

backend/app/db/models/uploaded_file.py

@@ -17,6 +17,7 @@ class FileTypeEnum(enum.Enum):
     ECMK_SITE_NOTE = "ecmk_site_note"
     ECMK_RD_SAP_SITE_NOTE = "ecmk_rd_sap_site_note"
     ECMK_SURVEY_XML = "ecmk_survey_xml"
+    MAGIC_PLAN_JSON = "magic_plan_json"
 
 
 class FileSourceEnum(enum.Enum):
@@ -24,6 +25,7 @@ class FileSourceEnum(enum.Enum):
     SHAREPOINT = "sharepoint"
     HUBSPOT = "hubspot"
     ECMK = "ecmk"
+    MAGIC_PLAN = "magic_plan"
 
 
 class UploadedFile(Base):

backend/magic_plan/handler.py

@@ -19,7 +19,8 @@ def handler(body: dict[str, Any], context: Any) -> str:
         customer_id=settings.MAGICPLAN_CUSTOMER_ID,
         api_key=settings.MAGICPLAN_API_KEY,
     )
-    plan: Plan = MagicPlanService(client).run(payload.address, payload.uprn)
+    # TODO: read s3_bucket from env var so staging/prod use the correct bucket
+    plan: Plan = MagicPlanService(client, s3_bucket="retrofit-energy-assessments-dev").run(payload)
     logger.info("Saved MagicPlan plan uid=%s", plan.uid)
     return plan.uid
 
@@ -28,7 +29,7 @@ if __name__ == "__main__":
     event = {
         "Records": [
             {
-                "body": '{"address": "2 Laburnum Way Bromley BR2 8BZ"}',
+                "body": '{"address": "2 Laburnum Way Bromley BR2 8BZ", "hubspot_deal_id": "local-test-deal"}',
                 "messageId": "local-test",
             }
         ]

backend/magic_plan/handler/Dockerfile

@@ -0,0 +1,12 @@
+FROM public.ecr.aws/lambda/python:3.11
+
+WORKDIR /var/task
+
+COPY backend/magic_plan/handler/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY utils/ utils/
+COPY backend/ backend/
+COPY datatypes/ datatypes/
+
+CMD ["backend.magic_plan.handler.handler"]

backend/magic_plan/handler/requirements.txt

@@ -0,0 +1,7 @@
+awslambdaric
+requests
+sqlalchemy==2.0.36
+sqlmodel
+psycopg2-binary==2.9.10
+pydantic-settings==2.6.0
+boto3==1.35.44

backend/magic_plan/magic_plan_client.py

@@ -17,8 +17,14 @@ class MagicPlanClient:
         return PlansListResponse.model_validate(r.json()["data"])
 
     def get_plan(self, plan_id: str) -> MagicPlanPlan:
+        return MagicPlanPlan.model_validate(self._fetch_plan(plan_id).json()["data"])
+
+    def get_plan_raw(self, plan_id: str) -> bytes:
+        return self._fetch_plan(plan_id).content
+
+    def _fetch_plan(self, plan_id: str) -> requests.Response:
         r = self._session.get(
             f"{_BASE_URL}/plans/{plan_id}", params={"key": self._api_key}
         )
         r.raise_for_status()
-        return MagicPlanPlan.model_validate(r.json()["data"])
+        return r

backend/magic_plan/magic_plan_service.py

@@ -1,3 +1,6 @@
+import gzip
+import json
+from datetime import datetime, timezone
 from typing import Optional
 
 from datatypes.magicplan.api.response import (
@@ -10,33 +13,78 @@ from datatypes.magicplan.domain.models import Plan
 
 from backend.app.db.connection import db_session
 from backend.app.db.functions.magic_plan_functions import save_plan
+from backend.app.db.models.uploaded_file import (
+    FileSourceEnum,
+    FileTypeEnum,
+    UploadedFile,
+)
 from backend.magic_plan.address_matcher import find_matching_plan
 from backend.magic_plan.magic_plan_client import MagicPlanClient
+from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
 from utils.logger import setup_logger
+from utils.s3 import save_data_to_s3
 
 logger = setup_logger()
 
 
 class MagicPlanService:
-    def __init__(self, client: MagicPlanClient) -> None:
+    def __init__(self, client: MagicPlanClient, s3_bucket: str) -> None:
         self._client = client
+        self._s3_bucket = s3_bucket
+
+    def run(self, request: MagicPlanTriggerRequest) -> Plan:
+        address = request.address
+        uprn = request.uprn
 
-    def run(self, address: str, uprn: Optional[str] = None) -> Plan:
         if uprn is not None:
             logger.info("MagicPlanService.run uprn=%s", uprn)
 
         plans_response: PlansListResponse = self._client.get_plans()
         matched: Optional[PlanSummary] = find_matching_plan(
             plans_response.plans, address
-        )  # TODO: use address2UPRN instead? or create AddressMatch domain class
+        )
 
         if matched is None:
             raise ValueError(f"No MagicPlan found for address: {address!r}")
 
-        magic_plan: MagicPlanPlan = self._client.get_plan(matched.id)
+        raw_bytes: bytes = self._client.get_plan_raw(matched.id)
+        magic_plan: MagicPlanPlan = MagicPlanPlan.model_validate(
+            json.loads(raw_bytes)["data"]
+        )
         plan: Plan = map_plan(magic_plan)
 
+        uploaded_file: UploadedFile = self._upload_raw_plan_json(
+            plan_id=matched.id,
+            raw_bytes=raw_bytes,
+            uprn=uprn,
+            hubspot_deal_id=request.hubspot_deal_id,
+        )
+
         with db_session() as session:
             save_plan(session, plan)
+            session.add(uploaded_file)
 
         return plan
+
+    def _upload_raw_plan_json(
+        self,
+        plan_id: str,
+        raw_bytes: bytes,
+        uprn: Optional[str],
+        hubspot_deal_id: str,
+    ) -> UploadedFile:
+        compressed = gzip.compress(raw_bytes)
+        if uprn is not None:
+            s3_key = f"documents/uprn/{uprn}/magic_plan_{plan_id}.json.gz"
+        else:
+            s3_key = f"documents/hubspot_deal_id/{hubspot_deal_id}/magic_plan_{plan_id}.json.gz"
+        save_data_to_s3(compressed, self._s3_bucket, s3_key)
+        return UploadedFile(
+            s3_file_bucket=self._s3_bucket,
+            s3_file_key=s3_key,
+            s3_upload_timestamp=datetime.now(timezone.utc),
+            uprn=int(uprn) if uprn is not None else None,
+            hubspot_deal_id=hubspot_deal_id,
+            file_source=FileSourceEnum.MAGIC_PLAN.value,
+            file_type=FileTypeEnum.MAGIC_PLAN_JSON.value,
+        )

backend/magic_plan/magic_plan_trigger_request.py

@@ -7,4 +7,5 @@ class MagicPlanTriggerRequest(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
     address: str
+    hubspot_deal_id: str
     uprn: Optional[str] = None

backend/magic_plan/tests/test_magic_plan_client.py

@@ -172,3 +172,55 @@ def test_get_plan_propagates_http_error(
     # Act / Assert
     with pytest.raises(requests.HTTPError):
         client.get_plan("some-id")
+
+
+# --- get_plan_raw ---
+
+
+def test_get_plan_raw_returns_bytes(
+    client: MagicPlanClient, mock_session: MagicMock
+) -> None:
+    # Arrange
+    mock_session.get.return_value.content = b'{"data": "raw"}'
+    plan_id = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
+    # Act
+    result = client.get_plan_raw(plan_id)
+    # Assert
+    assert isinstance(result, bytes)
+
+
+def test_get_plan_raw_calls_correct_url(
+    client: MagicPlanClient, mock_session: MagicMock
+) -> None:
+    # Arrange
+    mock_session.get.return_value.content = b"{}"
+    plan_id = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
+    # Act
+    client.get_plan_raw(plan_id)
+    # Assert
+    mock_session.get.assert_called_once_with(
+        f"{BASE_URL}/plans/{plan_id}", params={"key": API_KEY}
+    )
+
+
+def test_get_plan_raw_calls_raise_for_status(
+    client: MagicPlanClient, mock_session: MagicMock
+) -> None:
+    # Arrange
+    mock_session.get.return_value.content = b"{}"
+    # Act
+    client.get_plan_raw("a7285ed1-878d-47eb-8aa6-85ef9e187516")
+    # Assert
+    mock_session.get.return_value.raise_for_status.assert_called_once()
+
+
+def test_get_plan_raw_propagates_http_error(
+    client: MagicPlanClient, mock_session: MagicMock
+) -> None:
+    # Arrange
+    mock_session.get.return_value.raise_for_status.side_effect = requests.HTTPError(
+        "500"
+    )
+    # Act / Assert
+    with pytest.raises(requests.HTTPError):
+        client.get_plan_raw("some-id")

backend/magic_plan/tests/test_magic_plan_service.py

@@ -1,6 +1,6 @@
 import json
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import pytest
 
@@ -8,11 +8,18 @@ from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
 from datatypes.magicplan.domain.mapper import map_plan
 from datatypes.magicplan.domain.models import Plan
 
+from backend.app.db.models.uploaded_file import (
+    FileSourceEnum,
+    FileTypeEnum,
+    UploadedFile,
+)
 from backend.magic_plan.magic_plan_client import MagicPlanClient
 from backend.magic_plan.magic_plan_service import MagicPlanService
+from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
 
 FIXTURE_DIR = Path(__file__).parents[2] / "magic_plan"
 PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
+S3_BUCKET = "test-bucket"
 
 
 @pytest.fixture(scope="module")
@@ -41,11 +48,25 @@ def plan_summary() -> PlanSummary:
 
 @pytest.fixture()
 def mock_client() -> MagicMock:
-    return MagicMock(spec=MagicPlanClient)
+    client = MagicMock(spec=MagicPlanClient)
+    client.get_plan_raw.return_value = (
+        FIXTURE_DIR / "magicplan_api_plan_response_example.json"
+    ).read_bytes()
+    return client
 
 
 def _make_service(mock_client: MagicMock) -> MagicPlanService:
-    return MagicPlanService(client=mock_client)
+    return MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
+
+
+def _make_request(
+    address: str = "2 Laburnum Way Bromley BR2 8BZ",
+    hubspot_deal_id: str = "deal-123",
+    uprn: str | None = None,
+) -> MagicPlanTriggerRequest:
+    return MagicPlanTriggerRequest(
+        address=address, hubspot_deal_id=hubspot_deal_id, uprn=uprn
+    )
 
 
 # --- no match ---
@@ -57,7 +78,7 @@ def test_run_raises_when_no_plan_found(mock_client: MagicMock) -> None:
     service = _make_service(mock_client)
     # Act / Assert
     with pytest.raises(ValueError, match="No MagicPlan found"):
-        service.run("99 Nowhere Road London SW1A 1AA")
+        service.run(_make_request(address="99 Nowhere Road London SW1A 1AA"))
 
 
 # --- match found ---
@@ -78,10 +99,12 @@ def test_run_fetches_plan_with_matched_id(
         return_value=plan_summary,
     ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
         "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
     ):
-        service.run("2 Laburnum Way Bromley BR2 8BZ")
+        service.run(_make_request())
     # Assert
-    mock_client.get_plan.assert_called_once_with(plan_summary.id)
+    mock_client.get_plan_raw.assert_called_once_with(plan_summary.id)
 
 
 def test_run_returns_mapped_plan(
@@ -99,8 +122,10 @@ def test_run_returns_mapped_plan(
         return_value=plan_summary,
     ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
         "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
     ):
-        result = service.run("2 Laburnum Way Bromley BR2 8BZ")
+        result = service.run(_make_request())
     # Assert
     assert isinstance(result, Plan)
     assert result.uid == PLAN_ID
@@ -120,8 +145,10 @@ def test_run_calls_save_plan_with_mapped_plan(
         return_value=plan_summary,
     ), patch("backend.magic_plan.magic_plan_service.save_plan") as mock_save, patch(
         "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
     ):
-        service.run("2 Laburnum Way Bromley BR2 8BZ")
+        service.run(_make_request())
     # Assert — save_plan called with a Plan whose uid matches
     call_args = mock_save.call_args
     saved_plan: Plan = call_args[0][1]
@@ -142,5 +169,105 @@ def test_run_accepts_uprn_without_error(
         return_value=plan_summary,
     ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
         "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
     ):
-        service.run("2 Laburnum Way Bromley BR2 8BZ", uprn="100023336956")
+        service.run(_make_request(uprn="100023336956"))
+
+
+# --- S3 upload ---
+
+
+def test_run_uploads_to_s3_with_uprn_key(
+    mock_client: MagicMock,
+    api_magic_plan: MagicPlanPlan,
+    plan_summary: PlanSummary,
+) -> None:
+    # Arrange
+    mock_client.get_plans.return_value.plans = [plan_summary]
+    request = _make_request(uprn="100023336956")
+    service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
+    with patch(
+        "backend.magic_plan.magic_plan_service.find_matching_plan",
+        return_value=plan_summary,
+    ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
+        "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
+    ) as mock_s3:
+        # Act
+        service.run(request)
+    # Assert
+    mock_s3.assert_called_once_with(
+        ANY,
+        S3_BUCKET,
+        f"documents/uprn/100023336956/magic_plan_{plan_summary.id}.json.gz",
+    )
+
+
+def test_run_uploads_to_s3_with_deal_id_key_when_uprn_absent(
+    mock_client: MagicMock,
+    api_magic_plan: MagicPlanPlan,
+    plan_summary: PlanSummary,
+) -> None:
+    # Arrange
+    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plan.return_value = api_magic_plan
+    request = _make_request(hubspot_deal_id="deal-456", uprn=None)
+    service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
+    with patch(
+        "backend.magic_plan.magic_plan_service.find_matching_plan",
+        return_value=plan_summary,
+    ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
+        "backend.magic_plan.magic_plan_service.db_session"
+    ), patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
+    ) as mock_s3:
+        # Act
+        service.run(request)
+    # Assert
+    mock_s3.assert_called_once_with(
+        ANY,
+        S3_BUCKET,
+        f"documents/hubspot_deal_id/deal-456/magic_plan_{plan_summary.id}.json.gz",
+    )
+
+
+# --- UploadedFile record ---
+
+
+def test_run_creates_uploaded_file_record(
+    mock_client: MagicMock,
+    api_magic_plan: MagicPlanPlan,
+    plan_summary: PlanSummary,
+) -> None:
+    # Arrange
+    mock_client.get_plans.return_value.plans = [plan_summary]
+    mock_client.get_plan.return_value = api_magic_plan
+    request = _make_request(hubspot_deal_id="deal-789", uprn="100023336956")
+    service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
+    mock_session = MagicMock()
+    with patch(
+        "backend.magic_plan.magic_plan_service.find_matching_plan",
+        return_value=plan_summary,
+    ), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
+        "backend.magic_plan.magic_plan_service.db_session"
+    ) as mock_db, patch(
+        "backend.magic_plan.magic_plan_service.save_data_to_s3"
+    ):
+        mock_db.return_value.__enter__.return_value = mock_session
+        # Act
+        service.run(request)
+    # Assert
+    added_objects = [call.args[0] for call in mock_session.add.call_args_list]
+    uploaded_file = next(
+        (obj for obj in added_objects if isinstance(obj, UploadedFile)), None
+    )
+    assert uploaded_file is not None
+    assert uploaded_file.file_source == FileSourceEnum.MAGIC_PLAN.value
+    assert uploaded_file.file_type == FileTypeEnum.MAGIC_PLAN_JSON.value
+    assert uploaded_file.s3_file_bucket == S3_BUCKET
+    assert uploaded_file.s3_file_key == f"documents/uprn/100023336956/magic_plan_{plan_summary.id}.json.gz"
+    assert uploaded_file.s3_upload_timestamp is not None
+    assert uploaded_file.uprn == 100023336956
+    assert uploaded_file.hubspot_deal_id == "deal-789"

backend/magic_plan/tests/test_magic_plan_trigger_request.py

@@ -6,17 +6,18 @@ from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerReques
 
 def test_valid_payload_with_address_only() -> None:
     # Arrange
-    payload = {"address": "123 High St London SW1A 1AA"}
+    payload = {"address": "123 High St London SW1A 1AA", "hubspot_deal_id": "123456789"}
     # Act
     req = MagicPlanTriggerRequest.model_validate(payload)
     # Assert
     assert req.address == "123 High St London SW1A 1AA"
+    assert req.hubspot_deal_id == "123456789"
     assert req.uprn is None
 
 
 def test_valid_payload_with_uprn() -> None:
     # Arrange
-    payload = {"address": "123 High St London SW1A 1AA", "uprn": "100023336956"}
+    payload = {"address": "123 High St London SW1A 1AA", "hubspot_deal_id": "123456789", "uprn": "100023336956"}
     # Act
     req = MagicPlanTriggerRequest.model_validate(payload)
     # Assert
@@ -25,7 +26,7 @@ def test_valid_payload_with_uprn() -> None:
 
 def test_missing_address_raises() -> None:
     # Arrange
-    payload = {"uprn": "100023336956"}
+    payload = {"hubspot_deal_id": "123456789", "uprn": "100023336956"}
     # Act / Assert
     with pytest.raises(ValidationError):
         MagicPlanTriggerRequest.model_validate(payload)
@@ -33,8 +34,16 @@ def test_missing_address_raises() -> None:
 
 def test_extra_fields_ignored() -> None:
     # Arrange
-    payload = {"address": "123 High St London SW1A 1AA", "unknown_field": "whatever"}
+    payload = {"address": "123 High St London SW1A 1AA", "hubspot_deal_id": "123456789", "unknown_field": "whatever"}
     # Act
     req = MagicPlanTriggerRequest.model_validate(payload)
     # Assert
     assert req.address == "123 High St London SW1A 1AA"
+
+
+def test_missing_hubspot_deal_id_raises() -> None:
+    # Arrange
+    payload = {"address": "123 High St London SW1A 1AA"}
+    # Act / Assert
+    with pytest.raises(ValidationError):
+        MagicPlanTriggerRequest.model_validate(payload)

etl/hubspot/hubspot_deal_differ.py

@@ -162,6 +162,14 @@ class HubspotDealDiffer:
 
         return False
 
+    @staticmethod
+    def check_for_magicplan_trigger(
+        new_deal: Dict[str, str], old_deal: HubspotDealData
+    ) -> bool:
+        new_outcome = (new_deal.get("outcome") or "").lower()
+        old_outcome = (old_deal.outcome or "").lower()
+        return new_outcome == "surveyed" and old_outcome != "surveyed"
+
     @staticmethod
     def _has_valid_pashub_link(new_pashub_link: str) -> bool:
         return bool(new_pashub_link)
@@ -178,7 +186,7 @@ class HubspotDealDiffer:
     def _coordination_completed(
         new_deal: Dict[str, str], old_deal: HubspotDealData
     ) -> bool:
-        new_status: str = new_deal.get("coordination_status") or ""
+        new_status: str = new_deal.get("coordination_status__stage_1_") or ""
         return (
             new_status != ""
             and new_status.lower() in HubspotDealDiffer.COORDINATION_COMPLETE
@@ -187,7 +195,7 @@ class HubspotDealDiffer:
 
     @staticmethod
     def _design_completed(new_deal: Dict[str, str], old_deal: HubspotDealData) -> bool:
-        new_status: str = new_deal.get("design_status") or ""
+        new_status: str = new_deal.get("retrofit_design_status") or ""
         return (
             new_status != ""
             and new_status.lower() == HubspotDealDiffer.RETROFIT_DESIGN_COMPLETE

etl/hubspot/scripts/scraper/main.py

@@ -56,6 +56,12 @@ def handler(body: dict[str, Any], context: Any) -> None:
                 f"Triggering Pas Hub file fetcher for HubSpot deal ID {hubspot_deal_id}"
             )
             _trigger_pashub_fetcher(sqs_client, hubspot_deal_id, hubspot_deal)
+
+        if (hubspot_deal.get("outcome") or "").lower() == "surveyed":
+            logger.info(
+                f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}"
+            )
+            _trigger_magicplan_fetcher(sqs_client, hubspot_deal, listing, hubspot_deal_id)
     else:
         # Deal already in db, check whether anything has changed
         logger.info(
@@ -97,9 +103,34 @@ def handler(body: dict[str, Any], context: Any) -> None:
                 f"Not Triggering PasHub file fetcher for HubSpot deal ID {hubspot_deal_id}"
             )
 
+        if HubspotDealDiffer.check_for_magicplan_trigger(
+            new_deal=hubspot_deal, old_deal=db_deal
+        ):
+            logger.info(
+                f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}"
+            )
+            _trigger_magicplan_fetcher(sqs_client, hubspot_deal, listing, hubspot_deal_id)
+
     print("done")
 
 
+def _trigger_magicplan_fetcher(
+    sqs_client: Any, hubspot_deal: Dict[str, str], listing: Optional[dict[str, str]], hubspot_deal_id: str
+) -> None:
+    message_body = {
+        "address": hubspot_deal.get("dealname"),
+        "hubspot_deal_id": hubspot_deal_id,
+        "uprn": listing.get("national_uprn") if listing else None,
+    }
+    response = sqs_client.send_message(
+        QueueUrl=get_settings().MAGICPLAN_SQS_URL,
+        MessageBody=json.dumps(message_body),
+    )
+    logger.info(
+        f"Sent message to MagicPlan queue. MessageId: {response['MessageId']}"
+    )
+
+
 def _trigger_pashub_fetcher(
     sqs_client: Any, deal_id: str, hubspot_deal: Dict[str, str]
 ) -> None:

etl/hubspot/tests/test_hubspot_deal_differ.py

@@ -109,7 +109,7 @@ def test_pashub_trigger__coordination_completed_and_pashub_link_set__returns_tru
     new_deal = make_new_deal(
         deal_id,
         pashub_link="www.google.co.uk",
-        coordination_status=coordination_status,
+        **{"coordination_status__stage_1_": coordination_status},
     )
 
     assert (
@@ -156,7 +156,7 @@ def test_pashub_trigger__design_completed_and_pashub_link_set__returns_true() ->
     new_deal = make_new_deal(
         deal_id,
         pashub_link="www.google.co.uk",
-        design_status="uploaded",
+        retrofit_design_status="uploaded",
     )
 
     assert (
@@ -177,7 +177,7 @@ def test_pashub_trigger__design_completed_and_pashub_link_not_set__returns_false
 
     new_deal = make_new_deal(
         deal_id,
-        design_status="uploaded",
+        retrofit_design_status="uploaded",
     )
 
     assert (
@@ -270,6 +270,79 @@ def test_pashub_trigger__coordination_design_lodgement_not_completed_and_pashub_
     )
 
 
+# ==========================
+# MAGICPLAN TRIGGER TESTS
+# ==========================
+
+
+def test_magicplan_trigger__outcome_transitions_to_surveyed__returns_true() -> None:
+    deal_id = uuid.uuid4()
+
+    # Arrange
+    old_deal = make_old_deal(id=deal_id, outcome="assessed")
+    new_deal = make_new_deal(deal_id, outcome="surveyed")
+
+    # Act
+    result = HubspotDealDiffer.check_for_magicplan_trigger(
+        new_deal=new_deal,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is True
+
+
+def test_magicplan_trigger__outcome_already_surveyed__returns_false() -> None:
+    deal_id = uuid.uuid4()
+
+    # Arrange
+    old_deal = make_old_deal(id=deal_id, outcome="surveyed")
+    new_deal = make_new_deal(deal_id, outcome="surveyed")
+
+    # Act
+    result = HubspotDealDiffer.check_for_magicplan_trigger(
+        new_deal=new_deal,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is False
+
+
+def test_magicplan_trigger__outcome_transitions_to_non_surveyed__returns_false() -> None:
+    deal_id = uuid.uuid4()
+
+    # Arrange
+    old_deal = make_old_deal(id=deal_id, outcome="assessed")
+    new_deal = make_new_deal(deal_id, outcome="assessed")
+
+    # Act
+    result = HubspotDealDiffer.check_for_magicplan_trigger(
+        new_deal=new_deal,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is False
+
+
+def test_magicplan_trigger__outcome_surveyed_uppercase__returns_true() -> None:
+    deal_id = uuid.uuid4()
+
+    # Arrange
+    old_deal = make_old_deal(id=deal_id, outcome="assessed")
+    new_deal = make_new_deal(deal_id, outcome="SURVEYED")
+
+    # Act
+    result = HubspotDealDiffer.check_for_magicplan_trigger(
+        new_deal=new_deal,
+        old_deal=old_deal,
+    )
+
+    # Assert
+    assert result is True
+
+
 # =======================
 # DB UPDATE TRIGGER TESTS
 # =======================

etl/hubspot/tests/test_scraper_handler.py

@@ -0,0 +1,227 @@
+import json
+import uuid
+from typing import Any, Dict, Optional
+from unittest.mock import MagicMock, patch
+
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
+from etl.hubspot.scripts.scraper.main import handler
+
+DEAL_NAME = "123 Main Street"
+UPRN = "12345678"
+DEAL_ID = "999"
+PASHUB_LINK = "https://pashub.example.com/deal/999"
+MAGICPLAN_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123/magic-plan-dev"
+PASHUB_QUEUE_URL = "https://sqs.test/pashub"
+
+
+def make_hubspot_deal(**kwargs: Any) -> Dict[str, Any]:
+    return {
+        "hs_object_id": DEAL_ID,
+        "dealname": DEAL_NAME,
+        "pashub_link": None,
+        **kwargs,
+    }
+
+
+def make_db_deal(**kwargs: Any) -> HubspotDealData:
+    return HubspotDealData(
+        id=uuid.uuid4(),
+        deal_id=DEAL_ID,
+        **kwargs,
+    )
+
+
+def run_handler(
+    hubspot_deal: Dict[str, Any],
+    db_deal: Optional[HubspotDealData],
+    listing: Optional[dict],
+) -> MagicMock:
+    mock_sqs = MagicMock()
+    mock_sqs.send_message.return_value = {"MessageId": "test-id"}
+
+    with (
+        patch("etl.hubspot.scripts.scraper.main.HubspotDataToDb") as mock_db_cls,
+        patch("etl.hubspot.scripts.scraper.main.HubspotClient") as mock_hs_cls,
+        patch("etl.hubspot.scripts.scraper.main.boto3") as mock_boto3,
+        patch("etl.hubspot.scripts.scraper.main.get_settings") as mock_settings,
+    ):
+        mock_db_cls.return_value.find_deal_with_deal_id.return_value = db_deal
+        mock_db_cls.return_value.upsert_deal.return_value = None
+        mock_hs_cls.return_value.get_deal_and_company_and_listing.return_value = (
+            hubspot_deal,
+            None,
+            listing,
+        )
+        mock_boto3.client.return_value = mock_sqs
+        mock_settings.return_value.MAGICPLAN_SQS_URL = MAGICPLAN_QUEUE_URL
+        mock_settings.return_value.PASHUB_TO_ARA_SQS_URL = PASHUB_QUEUE_URL
+
+        handler.__wrapped__({"hubspot_deal_id": DEAL_ID}, "")
+
+    return mock_sqs
+
+
+# ====================================
+# NEW DEAL PATH - MagicPlan trigger
+# ====================================
+
+
+def test_new_deal_surveyed__sends_magicplan_sqs() -> None:
+    # Arrange
+    hubspot_deal = make_hubspot_deal(outcome="surveyed")
+    listing = {"national_uprn": UPRN}
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=listing)
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=MAGICPLAN_QUEUE_URL,
+        MessageBody=json.dumps(
+            {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": UPRN}
+        ),
+    )
+
+
+def test_new_deal_not_surveyed__no_magicplan_sqs() -> None:
+    # Arrange
+    hubspot_deal = make_hubspot_deal(outcome="assessed")
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_not_called()
+
+
+def test_new_deal_surveyed_no_listing__magicplan_sqs_uprn_is_null() -> None:
+    # Arrange
+    hubspot_deal = make_hubspot_deal(outcome="surveyed")
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=MAGICPLAN_QUEUE_URL,
+        MessageBody=json.dumps(
+            {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": None}
+        ),
+    )
+
+
+# ==========================================
+# EXISTING DEAL PATH - MagicPlan trigger
+# ==========================================
+
+
+def test_existing_deal_surveyed_transition__sends_magicplan_sqs() -> None:
+    # Arrange
+    db_deal = make_db_deal(outcome="assessed")
+    hubspot_deal = make_hubspot_deal(outcome="surveyed")
+    listing = {"national_uprn": UPRN}
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing)
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=MAGICPLAN_QUEUE_URL,
+        MessageBody=json.dumps(
+            {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": UPRN}
+        ),
+    )
+
+
+def test_existing_deal_already_surveyed__no_magicplan_sqs() -> None:
+    # Arrange
+    db_deal = make_db_deal(outcome="surveyed", dealname="Old Name")
+    hubspot_deal = make_hubspot_deal(outcome="surveyed", dealname="New Name")
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_not_called()
+
+
+# ====================================
+# NEW DEAL PATH - PasHub trigger
+# ====================================
+
+
+def test_new_deal_with_pashub_link__sends_pashub_sqs() -> None:
+    # Arrange
+    hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK)
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=PASHUB_QUEUE_URL,
+        MessageBody=json.dumps(
+            {
+                "pashub_link": PASHUB_LINK,
+                "address": None,
+                "hubspot_deal_id": DEAL_ID,
+                "sharepoint_link": None,
+                "uprn": None,
+                "landlord_property_id": None,
+                "deal_stage": None,
+            }
+        ),
+    )
+
+
+def test_new_deal_no_pashub_link__no_pashub_sqs() -> None:
+    # Arrange
+    hubspot_deal = make_hubspot_deal()
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_not_called()
+
+
+# ==========================================
+# EXISTING DEAL PATH - PasHub trigger
+# ==========================================
+
+
+def test_existing_deal_pashub_link_added__sends_pashub_sqs() -> None:
+    # Arrange
+    db_deal = make_db_deal(pashub_link=None)
+    hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK)
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_called_once_with(
+        QueueUrl=PASHUB_QUEUE_URL,
+        MessageBody=json.dumps(
+            {
+                "pashub_link": PASHUB_LINK,
+                "address": None,
+                "hubspot_deal_id": DEAL_ID,
+                "sharepoint_link": None,
+                "uprn": None,
+                "landlord_property_id": None,
+                "deal_stage": None,
+            }
+        ),
+    )
+
+
+def test_existing_deal_pashub_link_unchanged__no_pashub_sqs() -> None:
+    # Arrange
+    db_deal = make_db_deal(pashub_link=PASHUB_LINK, dealname="Old Name")
+    hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK, dealname="New Name")
+
+    # Act
+    mock_sqs = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None)
+
+    # Assert
+    mock_sqs.send_message.assert_not_called()

infrastructure/terraform/lambda/hubspot_deal_etl/main.tf

@@ -12,7 +12,16 @@ data "terraform_remote_state" "pashub_to_ara" {
   config = {
     bucket = "pashub-to-ara-terraform-state"
     key    = "env:/${var.stage}/terraform.tfstate"
-    region = "eu-west-2" 
+    region = "eu-west-2"
+  }
+}
+
+data "terraform_remote_state" "magic_plan" {
+  backend = "s3"
+  config = {
+    bucket = "magic-plan-hubspot-trigger-terraform-state"
+    key    = "env:/${var.stage}/terraform.tfstate"
+    region = "eu-west-2"
   }
 }
 
@@ -49,6 +58,7 @@ module "hubspot_deal_etl" {
     HUBSPOT_API_KEY = var.hubspot_api_key
 
     PASHUB_TO_ARA_SQS_URL = data.terraform_remote_state.pashub_to_ara.outputs.pashub_to_ara_queue_url
+    MAGICPLAN_SQS_URL     = data.terraform_remote_state.magic_plan.outputs.magic_plan_queue_url
   }
 }
 
@@ -76,4 +86,18 @@ module "hubspot_deal_etl_sqs_policy" {
 resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_sqs_send" {
   role       = module.hubspot_deal_etl.role_name
   policy_arn = module.hubspot_deal_etl_sqs_policy.policy_arn
+}
+
+module "hubspot_deal_etl_magicplan_sqs_policy" {
+  source = "../../modules/general_iam_policy"
+
+  policy_name        = "hubspot-deal-etl-magicplan-sqs-send-${var.stage}"
+  policy_description = "Allow HubSpot ETL Lambda to send messages to MagicPlan queue"
+  actions            = ["sqs:SendMessage"]
+  resources          = [data.terraform_remote_state.magic_plan.outputs.magic_plan_queue_arn]
+}
+
+resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_magicplan_sqs_send" {
+  role       = module.hubspot_deal_etl.role_name
+  policy_arn = module.hubspot_deal_etl_magicplan_sqs_policy.policy_arn
 }

infrastructure/terraform/lambda/magic_plan/main.tf

@@ -0,0 +1,46 @@
+data "terraform_remote_state" "shared" {
+  backend = "s3"
+  config = {
+    bucket = "assessment-model-terraform-state"
+    key    = "env:/${var.stage}/terraform.tfstate"
+    region = "eu-west-2"
+  }
+}
+
+data "aws_secretsmanager_secret_version" "db_credentials" {
+  secret_id = "${var.stage}/assessment_model/db_credentials"
+}
+
+locals {
+  db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string)
+}
+
+resource "aws_iam_role_policy_attachment" "magic_plan_s3_write" {
+  role       = module.lambda.role_name
+  policy_arn = data.terraform_remote_state.shared.outputs.energy_assessments_s3_write_arn
+}
+
+module "lambda" {
+  source = "../../modules/lambda_with_sqs"
+
+  name  = "magic_plan"
+  stage = var.stage
+
+  image_uri = local.image_uri
+
+  maximum_concurrency            = var.maximum_concurrency
+  reserved_concurrent_executions = var.reserved_concurrent_executions
+  batch_size                     = var.batch_size
+
+  environment = {
+    STAGE                 = var.stage
+    LOG_LEVEL             = "info"
+    MAGICPLAN_CUSTOMER_ID = var.magicplan_customer_id
+    MAGICPLAN_API_KEY     = var.magicplan_api_key
+    DB_USERNAME           = local.db_credentials.db_assessment_model_username
+    DB_PASSWORD           = local.db_credentials.db_assessment_model_password
+    DB_HOST               = var.db_host
+    DB_NAME               = var.db_name
+    DB_PORT               = var.db_port
+  }
+}

infrastructure/terraform/lambda/magic_plan/outputs.tf

@@ -0,0 +1,9 @@
+output "magic_plan_queue_url" {
+  value       = module.lambda.queue_url
+  description = "URL of the MagicPlan SQS queue"
+}
+
+output "magic_plan_queue_arn" {
+  value       = module.lambda.queue_arn
+  description = "ARN of the MagicPlan SQS queue"
+}

infrastructure/terraform/lambda/magic_plan/provider.tf

@@ -0,0 +1,16 @@
+terraform {
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = ">= 5.0"
+    }
+  }
+
+  backend "s3" {
+    bucket = "magic-plan-hubspot-trigger-terraform-state"
+    key    = "terraform.tfstate"
+    region = "eu-west-2"
+  }
+
+  required_version = ">= 1.2.0"
+}

infrastructure/terraform/lambda/magic_plan/variables.tf

@@ -0,0 +1,68 @@
+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 "maximum_concurrency" {
+  type        = number
+  default     = null
+  description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit."
+}
+
+variable "reserved_concurrent_executions" {
+  type    = number
+  default = 1
+}
+
+variable "batch_size" {
+  type    = number
+  default = 1
+}
+
+locals {
+  image_uri = "${var.ecr_repo_url}@${var.image_digest}"
+}
+
+output "resolved_image_uri" {
+  value = local.image_uri
+}
+
+variable "magicplan_customer_id" {
+  type      = string
+  sensitive = true
+}
+
+variable "magicplan_api_key" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_host" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_name" {
+  type      = string
+  sensitive = true
+}
+
+variable "db_port" {
+  type      = string
+  sensitive = true
+}

infrastructure/terraform/lambda/pashub_to_ara/main.tf

@@ -54,5 +54,5 @@ module "lambda" {
 
 resource "aws_iam_role_policy_attachment" "pashub_to_ara_s3_write" {
   role       = module.lambda.role_name
-  policy_arn = data.terraform_remote_state.shared.outputs.pashub_to_ara_s3_write_arn
+  policy_arn = data.terraform_remote_state.shared.outputs.energy_assessments_s3_write_arn
 }

infrastructure/terraform/shared/main.tf

@@ -568,18 +568,18 @@ module "pashub_to_ara_registry" {
   stage  = var.stage
 }
 
-module "pashub_to_ara_s3_write" {
+module "energy_assessments_s3_write" {
   source = "../modules/s3_iam_policy"
 
-  policy_name        = "PashubToAraWriteS3"
-  policy_description = "Allow PasHub to ARA Lambda to write to retrofit energy assessments bucket"
+  policy_name        = "EnergyAssessmentsWriteS3"
+  policy_description = "Allow lambdas to write to retrofit energy assessments bucket"
   bucket_arns        = ["arn:aws:s3:::retrofit-energy-assessments-${var.stage}"]
   actions            = ["s3:PutObject", "s3:AbortMultipartUpload"]
   resource_paths     = ["/*"]
 }
 
-output "pashub_to_ara_s3_write_arn" {
-  value = module.pashub_to_ara_s3_write.policy_arn
+output "energy_assessments_s3_write_arn" {
+  value = module.energy_assessments_s3_write.policy_arn
 }
 
 ################################################
@@ -745,4 +745,5 @@ module "magic_plan_client_registry" {
   source = "../modules/container_registry"
   name   = "magic-plan"
   stage  = var.stage
-}
+}
+

pytest.ini

@@ -3,6 +3,6 @@ pythonpath = .
 log_cli = true
 log_cli_level = INFO
 addopts = --cov-report term-missing --cov=etl/epc --cov=recommendations --cov=backend --cov=etl/epc_clean --cov=etl/spatial
-testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests backend/hubspot_trigger_orchestrator/tests datatypes/epc/schema/tests datatypes/epc/surveys/tests datatypes/epc/domain/tests backend/ecmk_fetcher/tests/ backend/pashub_fetcher/tests backend/documents_parser/tests backend/magic_plan/tests datatypes/magicplan/api/tests datatypes/magicplan/domain/tests backend/app/db/functions/tests
+testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests datatypes/epc/schema/tests datatypes/epc/surveys/tests datatypes/epc/domain/tests backend/ecmk_fetcher/tests/ backend/pashub_fetcher/tests backend/documents_parser/tests backend/magic_plan/tests datatypes/magicplan/api/tests datatypes/magicplan/domain/tests backend/app/db/functions/tests
 markers =
     integration: mark a test as an integration test