diff --git a/applications/abri/dispatch.py b/applications/abri/dispatch.py index aef12a3b2..f69981162 100644 --- a/applications/abri/dispatch.py +++ b/applications/abri/dispatch.py @@ -18,8 +18,10 @@ from orchestration.abri_orchestrator import ( ConfirmedSurveyBooking, DealDatabaseGateway, LogJobOrchestrationResult, + LogJobWriteBackError, TenantDataSyncResult, ) +from utilities.aws_lambda.task_handler import NonRetriableTaskError class AbriFlows(Protocol): @@ -56,4 +58,80 @@ def dispatch_abri_flows( deal_database: DealDatabaseGateway, ) -> Dict[str, str]: """Run the fired flows in the fixed order; returns a PII-free summary.""" - raise NotImplementedError + summary: Dict[str, str] = {} + + if "amend_job" in request.flows: + summary["amend_job"] = _run_amend(request, flows) + if "log_job" in request.flows: + summary["log_job"] = _run_log(request, flows, deal_database) + if "sync_tenant_data" in request.flows: + summary["sync_tenant_data"] = _run_tenant_sync(request, flows) + + return summary + + +def _run_amend(request: AbriTriggerRequest, flows: AbriFlows) -> str: + if request.confirmed_survey_date is None: + raise ValueError("amend_job fired without a confirmed_survey_date") + outcome = flows.amend_job( + AppointmentChange( + deal_id=request.hubspot_deal_id, + confirmed_survey_date=request.confirmed_survey_date, + confirmed_survey_time=request.confirmed_survey_time, + ) + ) + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="amend_job", rejection=outcome) + return ( + f"appointment amended to {outcome.appointment_date} " + f"{outcome.appointment_time} (job {outcome.job_no})" + ) + + +def _run_log( + request: AbriTriggerRequest, + flows: AbriFlows, + deal_database: DealDatabaseGateway, +) -> str: + existing_job_no = deal_database.job_no_for_deal(request.hubspot_deal_id) + if existing_job_no is not None: + return f"skipped: job {existing_job_no} already logged" + + if ( + request.place_ref is None + or request.deal_name is None + or request.confirmed_survey_date is None + ): + raise ValueError("log_job fired without place_ref/deal_name/survey date") + try: + outcome = flows.log_job( + ConfirmedSurveyBooking( + deal_id=request.hubspot_deal_id, + place_ref=PlaceRef(request.place_ref), + deal_name=request.deal_name, + confirmed_survey_date=request.confirmed_survey_date, + confirmed_survey_time=request.confirmed_survey_time, + ) + ) + except LogJobWriteBackError as error: + # The job exists in OpenHousing; a redelivery would log a duplicate. + # The orphaned job_no stays visible in the failure record. + raise NonRetriableTaskError(str(error)) from error + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="log_job", rejection=outcome) + return f"job {outcome.job_no} logged" + + +def _run_tenant_sync(request: AbriTriggerRequest, flows: AbriFlows) -> str: + if request.place_ref is None: + raise ValueError("sync_tenant_data fired without a place_ref") + outcome = flows.sync_tenant_data( + place_ref=PlaceRef(request.place_ref), + deal_id=request.hubspot_deal_id, + ) + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="sync_tenant_data", rejection=outcome) + return ( + f"{len(outcome.contact_ids)} contacts created " + f"({outcome.vulnerable_contact_count} vulnerable)" + )