From dd9592a1925addb845e8c335eede85dae8536ef0 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 29 Jul 2026 17:06:15 +0200 Subject: [PATCH] Refactor campaign delivery decision paths --- .../backend/persistence/campaigns.py | 337 +++++++++------- .../backend/persistence/versions.py | 370 ++++++++++++------ .../backend/reports/campaigns.py | 148 ++++--- src/govoplan_campaign/backend/router.py | 164 ++++++-- src/govoplan_campaign/backend/schemas.py | 101 ++--- src/govoplan_campaign/backend/sending/jobs.py | 308 +++++++++------ .../backend/sending/postbox_delivery.py | 147 ++++--- tests/test_postbox_delivery_orchestration.py | 111 ++++++ tests/test_report_email_security.py | 10 + 9 files changed, 1103 insertions(+), 593 deletions(-) diff --git a/src/govoplan_campaign/backend/persistence/campaigns.py b/src/govoplan_campaign/backend/persistence/campaigns.py index f101303..1dfd831 100644 --- a/src/govoplan_campaign/backend/persistence/campaigns.py +++ b/src/govoplan_campaign/backend/persistence/campaigns.py @@ -465,6 +465,177 @@ def _job_from_message( ) +def _resolve_built_postbox_targets( + session: Session, + *, + tenant_id: str, + config: CampaignConfig, + built_messages: list[Any], + entries_by_index: dict[int, Any], +) -> dict[int, list[dict[str, Any]]]: + resolved_by_index: dict[int, list[dict[str, Any]]] = {} + for built in built_messages: + if not DeliveryChannelPolicy(built.draft.delivery_channel_policy).uses_postbox: + continue + entry = entries_by_index.get(built.draft.entry_index) + if entry is None: + raise CampaignPersistenceError("Built recipient row is missing from the campaign input.") + resolved, issues, validation_status = resolve_entry_postbox_targets( + session, + tenant_id=tenant_id, + config=config, + entry=entry, + validation_status=built.draft.validation_status, + materialize=True, + ) + built.draft.issues.extend(issues) + built.draft.validation_status = validation_status + resolved_by_index[built.draft.entry_index] = resolved + return resolved_by_index + + +def _campaign_build_report(result: Any, files: Any) -> dict[str, Any]: + report_json = result.report.model_dump(mode="json", by_alias=True) + for message_payload, message in zip(report_json.get("messages", []), result.report.messages, strict=False): + if isinstance(message_payload, dict): + message_payload["attachments"] = [files.public_attachment_summary_payload(item) for item in message.attachments] + report_json.update({ + "built_at": datetime.now(UTC).isoformat(), + "build_token": uuid4().hex, + "built_count": result.report.built_count, + "build_failed_count": result.report.build_failed_count, + "ready_count": result.report.ready_count, + "warning_count": result.report.warning_count, + "needs_review_count": result.report.needs_review_count, + "blocked_count": result.report.blocked_count, + "excluded_count": result.report.excluded_count, + "inactive_count": result.report.inactive_count, + "queueable_count": result.report.queueable_count, + }) + return report_json + + +def _replace_version_jobs( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version_id: str, + built_messages: list[Any], + postbox_targets_by_index: dict[int, list[dict[str, Any]]], +) -> list[tuple[CampaignJob, MessageDraft]]: + session.query(CampaignIssue).filter( + CampaignIssue.campaign_version_id == version_id, + CampaignIssue.job_id.is_not(None), + ).delete(synchronize_session=False) + session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).delete(synchronize_session=False) + session.flush() + + pairs: list[tuple[CampaignJob, MessageDraft]] = [] + for built in built_messages: + job = _job_from_message( + tenant_id=tenant_id, + campaign_id=campaign_id, + version_id=version_id, + message=built.draft, + resolved_postbox_targets=postbox_targets_by_index.get(built.draft.entry_index, []), + ) + session.add(job) + pairs.append((job, built.draft)) + session.flush() + return pairs + + +def _mail_execution_profile( + session: Session, + *, + version: CampaignVersion, + config: CampaignConfig, + jobs: list[CampaignJob], +) -> tuple[str | None, dict[str, Any]]: + if not any(DeliveryChannelPolicy(job.delivery_channel_policy).uses_mail for job in jobs): + return None, {} + if not config.server.profile_capabilities.smtp_available: + raise CampaignPersistenceError("The selected Mail profile has no SMTP configuration; an execution snapshot cannot be created.") + profile_id = campaign_mail_profile_id(version.raw_json if isinstance(version.raw_json, dict) else {}) + if profile_id is None: + raise CampaignPersistenceError("Select an authorized Mail profile before building campaign messages that use Mail.") + summary = profile_delivery_summary(session, version) + if not summary.get("smtp_transport_revision"): + raise CampaignPersistenceError("The selected Mail profile has no SMTP transport revision.") + return profile_id, summary + + +def _store_execution_snapshot( + session: Session, + *, + version: CampaignVersion, + config: CampaignConfig, + jobs: list[CampaignJob], + build_summary: dict[str, Any], +) -> None: + profile_id, profile = _mail_execution_profile(session, version=version, config=config, jobs=jobs) + snapshot, snapshot_hash = create_execution_snapshot( + version, + mail_profile_id=profile_id, + smtp_server_id=profile.get("smtp_server_id"), + smtp_credential_id=profile.get("smtp_credential_id"), + imap_server_id=profile.get("imap_server_id"), + imap_credential_id=profile.get("imap_credential_id"), + smtp_transport_revision=profile.get("smtp_transport_revision"), + imap_transport_revision=profile.get("imap_transport_revision"), + delivery=config.delivery, + jobs=jobs, + build_summary=build_summary, + ) + version.execution_snapshot = snapshot + version.execution_snapshot_hash = snapshot_hash + version.execution_snapshot_at = datetime.now(UTC) + + +def _store_job_issues( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version_id: str, + job_build_pairs: list[tuple[CampaignJob, MessageDraft]], +) -> None: + for job, message in job_build_pairs: + session.add_all([ + CampaignIssue( + tenant_id=tenant_id, + campaign_id=campaign_id, + campaign_version_id=version_id, + job_id=job.id, + severity=issue.severity, + code=issue.code, + message=issue.message, + source=issue.source, + behavior=issue.behavior, + ) + for issue in message.issues + ]) + + +def _apply_campaign_build_state( + campaign: Campaign, + version: CampaignVersion, + *, + needs_review_count: int, + blocked_count: int, + queueable_count: int, +) -> None: + if needs_review_count or blocked_count: + campaign.status = CampaignStatus.NEEDS_REVIEW.value + version.workflow_state = CampaignVersionWorkflowState.APPROVED.value + elif queueable_count > 0: + campaign.status = CampaignStatus.READY_TO_QUEUE.value + version.workflow_state = CampaignVersionWorkflowState.BUILT.value + else: + campaign.status = CampaignStatus.VALIDATED.value + + def build_campaign_version( session: Session, *, @@ -508,154 +679,44 @@ def build_campaign_version( start=1, ) } - resolved_postbox_targets_by_index: dict[ - int, - list[dict[str, Any]], - ] = {} - for built in result.built_messages: - policy = DeliveryChannelPolicy( - built.draft.delivery_channel_policy - ) - if not policy.uses_postbox: - continue - entry = entries_by_index.get(built.draft.entry_index) - if entry is None: - raise CampaignPersistenceError( - "Built recipient row is missing from the campaign input." - ) - resolved_targets, postbox_issues, validation_status = ( - resolve_entry_postbox_targets( - session, - tenant_id=tenant_id, - config=managed_config, - entry=entry, - validation_status=built.draft.validation_status, - materialize=True, - ) - ) - built.draft.issues.extend(postbox_issues) - built.draft.validation_status = validation_status - resolved_postbox_targets_by_index[built.draft.entry_index] = ( - resolved_targets - ) - report_json = result.report.model_dump(mode="json", by_alias=True) - for message_payload, message in zip(report_json.get("messages", []), result.report.messages, strict=False): - if isinstance(message_payload, dict): - message_payload["attachments"] = [files.public_attachment_summary_payload(item) for item in message.attachments] - report_json["built_at"] = datetime.now(UTC).isoformat() - report_json["build_token"] = uuid4().hex - report_json.update({ - "built_count": result.report.built_count, - "build_failed_count": result.report.build_failed_count, - "ready_count": result.report.ready_count, - "warning_count": result.report.warning_count, - "needs_review_count": result.report.needs_review_count, - "blocked_count": result.report.blocked_count, - "excluded_count": result.report.excluded_count, - "inactive_count": result.report.inactive_count, - "queueable_count": result.report.queueable_count, - }) + resolved_postbox_targets_by_index = _resolve_built_postbox_targets( + session, + tenant_id=tenant_id, + config=managed_config, + built_messages=result.built_messages, + entries_by_index=entries_by_index, + ) + report_json = _campaign_build_report(result, files) version.build_summary = report_json editor_state = copy.deepcopy(version.editor_state or {}) editor_state.pop("review_send", None) version.editor_state = editor_state - # Rebuild jobs for the current version. Later, protect sent jobs from destructive rebuilds. - session.query(CampaignIssue).filter(CampaignIssue.campaign_version_id == version.id, CampaignIssue.job_id.is_not(None)).delete(synchronize_session=False) - session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version.id).delete(synchronize_session=False) - session.flush() - - job_build_pairs: list[tuple[CampaignJob, MessageDraft]] = [] - for built in result.built_messages: - job = _job_from_message( - tenant_id=tenant_id, - campaign_id=campaign.id, - version_id=version.id, - message=built.draft, - resolved_postbox_targets=resolved_postbox_targets_by_index.get( - built.draft.entry_index, - [], - ), - ) - session.add(job) - job_build_pairs.append((job, built.draft)) - - # Assign all job IDs in one round-trip, then persist exact attachment use - # records in bulk. This avoids one flush plus several metadata queries per - # recipient for large campaigns. - session.flush() - files.record_campaign_attachment_uses_for_jobs( + job_build_pairs = _replace_version_jobs( session, - [job for job, _message in job_build_pairs], - stage="built", + tenant_id=tenant_id, + campaign_id=campaign.id, + version_id=version.id, + built_messages=result.built_messages, + postbox_targets_by_index=resolved_postbox_targets_by_index, ) - uses_mail = any( - DeliveryChannelPolicy(job.delivery_channel_policy).uses_mail - for job, _message in job_build_pairs + jobs = [job for job, _message in job_build_pairs] + files.record_campaign_attachment_uses_for_jobs(session, jobs, stage="built") + _store_execution_snapshot(session, version=version, config=managed_config, jobs=jobs, build_summary=report_json) + _store_job_issues( + session, + tenant_id=tenant_id, + campaign_id=campaign.id, + version_id=version.id, + job_build_pairs=job_build_pairs, ) - profile_id: str | None = None - profile_summary: dict[str, Any] = {} - if uses_mail: - if not managed_config.server.profile_capabilities.smtp_available: - raise CampaignPersistenceError( - "The selected Mail profile has no SMTP configuration; an " - "execution snapshot cannot be created." - ) - profile_id = campaign_mail_profile_id( - version.raw_json if isinstance(version.raw_json, dict) else {} - ) - if profile_id is None: - raise CampaignPersistenceError( - "Select an authorized Mail profile before building campaign " - "messages that use Mail." - ) - profile_summary = profile_delivery_summary(session, version) - if not profile_summary.get("smtp_transport_revision"): - raise CampaignPersistenceError( - "The selected Mail profile has no SMTP transport revision." - ) - execution_snapshot, execution_snapshot_hash = create_execution_snapshot( + _apply_campaign_build_state( + campaign, version, - mail_profile_id=profile_id, - smtp_server_id=profile_summary.get("smtp_server_id"), - smtp_credential_id=profile_summary.get("smtp_credential_id"), - imap_server_id=profile_summary.get("imap_server_id"), - imap_credential_id=profile_summary.get("imap_credential_id"), - smtp_transport_revision=profile_summary.get( - "smtp_transport_revision" - ), - imap_transport_revision=profile_summary.get("imap_transport_revision"), - delivery=managed_config.delivery, - jobs=[job for job, _message in job_build_pairs], - build_summary=report_json, + needs_review_count=result.report.needs_review_count, + blocked_count=result.report.blocked_count, + queueable_count=result.report.queueable_count, ) - version.execution_snapshot = execution_snapshot - version.execution_snapshot_hash = execution_snapshot_hash - version.execution_snapshot_at = datetime.now(UTC) - for job, message in job_build_pairs: - for issue in message.issues: - session.add( - CampaignIssue( - tenant_id=tenant_id, - campaign_id=campaign.id, - campaign_version_id=version.id, - job_id=job.id, - severity=issue.severity, - code=issue.code, - message=issue.message, - source=issue.source, - behavior=issue.behavior, - ) - ) - - if result.report.needs_review_count or result.report.blocked_count: - campaign.status = CampaignStatus.NEEDS_REVIEW.value - version.workflow_state = CampaignVersionWorkflowState.APPROVED.value - elif result.report.queueable_count > 0: - campaign.status = CampaignStatus.READY_TO_QUEUE.value - version.workflow_state = CampaignVersionWorkflowState.BUILT.value - else: - campaign.status = CampaignStatus.VALIDATED.value session.add(version) session.add(campaign) diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index e708c83..2e0ca2c 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -328,6 +328,112 @@ def _apply_campaign_metadata(campaign: Campaign, raw_json: dict[str, Any]) -> No campaign.external_id = campaign_meta.get("id") or campaign.external_id +def _assert_fork_source_allowed(session: Session, *, campaign: Campaign, source: CampaignVersion) -> None: + if campaign_has_active_working_version(session, campaign): + current = session.get(CampaignVersion, campaign.current_version_id) + current_number = current.version_number if current else "current" + raise LockedCampaignVersionError( + f"Campaign already has active working version #{current_number}. " + "Unlock or continue editing that version instead of creating a parallel draft." + ) + if campaign.current_version_id and source.id != campaign.current_version_id: + raise LockedCampaignVersionError( + "Historical versions remain review-only and cannot become a new branch. " + "Create the next working copy from the campaign's current immutable version." + ) + + +def _fork_runtime_json( + session: Session, + *, + tenant_id: str, + campaign: Campaign, + source: CampaignVersion, + raw_json: dict[str, Any] | None, + source_filename: str | None, + source_base_path: str | None, + migrate_legacy_mail_settings: bool, +) -> dict[str, Any]: + source_json = source.raw_json if isinstance(source.raw_json, dict) else {} + requires_migration = bool(campaign_mail_profile_boundary_violations(source_json)) + if requires_migration and not migrate_legacy_mail_settings: + raise CampaignPersistenceError( + "This version contains legacy campaign-local SMTP/IMAP settings. Create the editable copy from " + "the Mail settings migration action so the audit record is preserved and the copy uses a Mail profile." + ) + base_json = raw_json if raw_json is not None else copy.deepcopy(source_json) + if requires_migration and raw_json is None: + base_json["server"] = public_campaign_mail_server(source_json) + assert_server_safe_campaign_paths( + base_json, + source_filename=source_filename, + source_base_path=source_base_path, + managed_files_available=files_integration().available, + ) + runtime_json = normalize_campaign_paths(base_json, source_base_path) if source_base_path else copy.deepcopy(base_json) + assert_campaign_uses_mail_profile_reference(runtime_json) + mail_integration().assert_campaign_mail_policy_allows_json( + session, + tenant_id=tenant_id, + raw_json=runtime_json, + campaign_id=campaign.id, + ) + return runtime_json + + +def _new_forked_campaign_version( + session: Session, + *, + campaign: Campaign, + source: CampaignVersion, + runtime_json: dict[str, Any], + current_flow: str | None, + current_step: str | None, + editor_state: dict[str, Any] | None, + source_filename: str | None, + source_base_path: str | None, + autosave: bool, +) -> CampaignVersion: + return CampaignVersion( + campaign_id=campaign.id, + version_number=_next_version_number(session, campaign.id), + raw_json=runtime_json, + schema_version=str(runtime_json.get("version", source.schema_version or "1.0")), + source_filename=source_filename if source_filename is not None else source.source_filename, + source_base_path=source_base_path if source_base_path is not None else source.source_base_path, + workflow_state=CampaignVersionWorkflowState.EDITING.value, + current_flow=current_flow if current_flow is not None else (source.current_flow or CampaignVersionFlow.MANUAL.value), + current_step=current_step if current_step is not None else source.current_step, + is_complete=False, + editor_state=( + validate_campaign_editor_state(editor_state) + if editor_state is not None + else campaign_editor_state_for_edit(source.editor_state) + ), + autosaved_at=datetime.now(UTC) if autosave else None, + ) + + +def _persist_forked_version( + session: Session, + *, + campaign: Campaign, + version: CampaignVersion, + commit: bool, +) -> None: + session.add(version) + session.flush() + _apply_campaign_metadata(campaign, version.raw_json) + campaign.current_version_id = version.id + campaign.status = CampaignStatus.DRAFT.value + session.add(campaign) + if commit: + _write_campaign_snapshot(version) + session.commit() + else: + session.flush() + + def fork_campaign_version_for_edit( session: Session, *, @@ -354,69 +460,30 @@ def fork_campaign_version_for_edit( source = get_campaign_version_for_tenant(session, tenant_id=tenant_id, campaign_id=campaign_id, version_id=version_id) campaign = _require_campaign(session, campaign_id) - if campaign_has_active_working_version(session, campaign): - current = session.get(CampaignVersion, campaign.current_version_id) - current_number = current.version_number if current else "current" - raise LockedCampaignVersionError( - f"Campaign already has active working version #{current_number}. " - "Unlock or continue editing that version instead of creating a parallel draft." - ) - if campaign.current_version_id and source.id != campaign.current_version_id: - raise LockedCampaignVersionError( - "Historical versions remain review-only and cannot become a new branch. " - "Create the next working copy from the campaign's current immutable version." - ) - - source_json = source.raw_json if isinstance(source.raw_json, dict) else {} - source_requires_mail_migration = bool(campaign_mail_profile_boundary_violations(source_json)) - if source_requires_mail_migration and not migrate_legacy_mail_settings: - raise CampaignPersistenceError( - "This version contains legacy campaign-local SMTP/IMAP settings. Create the editable copy from " - "the Mail settings migration action so the audit record is preserved and the copy uses a Mail profile." - ) - base_json = raw_json if raw_json is not None else copy.deepcopy(source_json) - if source_requires_mail_migration and raw_json is None: - base_json["server"] = public_campaign_mail_server(source_json) - assert_server_safe_campaign_paths( - base_json, + _assert_fork_source_allowed(session, campaign=campaign, source=source) + runtime_json = _fork_runtime_json( + session, + tenant_id=tenant_id, + campaign=campaign, + source=source, + raw_json=raw_json, source_filename=source_filename, source_base_path=source_base_path, - managed_files_available=files_integration().available, + migrate_legacy_mail_settings=migrate_legacy_mail_settings, ) - runtime_json = normalize_campaign_paths(base_json, source_base_path) if source_base_path else copy.deepcopy(base_json) - assert_campaign_uses_mail_profile_reference(runtime_json) - mail_integration().assert_campaign_mail_policy_allows_json(session, tenant_id=tenant_id, raw_json=runtime_json, campaign_id=campaign.id) - - new_version = CampaignVersion( - campaign_id=campaign.id, - version_number=_next_version_number(session, campaign.id), - raw_json=runtime_json, - schema_version=str(runtime_json.get("version", source.schema_version or "1.0")), - source_filename=source_filename if source_filename is not None else source.source_filename, - source_base_path=source_base_path if source_base_path is not None else source.source_base_path, - workflow_state=CampaignVersionWorkflowState.EDITING.value, - current_flow=current_flow if current_flow is not None else (source.current_flow or CampaignVersionFlow.MANUAL.value), - current_step=current_step if current_step is not None else source.current_step, - is_complete=False, - editor_state=( - validate_campaign_editor_state(editor_state) - if editor_state is not None - else campaign_editor_state_for_edit(source.editor_state) - ), - autosaved_at=datetime.now(UTC) if autosave else None, + new_version = _new_forked_campaign_version( + session, + campaign=campaign, + source=source, + runtime_json=runtime_json, + current_flow=current_flow, + current_step=current_step, + editor_state=editor_state, + source_filename=source_filename, + source_base_path=source_base_path, + autosave=autosave, ) - session.add(new_version) - session.flush() - - _apply_campaign_metadata(campaign, runtime_json) - campaign.current_version_id = new_version.id - campaign.status = CampaignStatus.DRAFT.value - session.add(campaign) - if commit: - _write_campaign_snapshot(new_version) - session.commit() - else: - session.flush() + _persist_forked_version(session, campaign=campaign, version=new_version, commit=commit) return new_version @@ -530,6 +597,113 @@ def unlock_validated_campaign_version( session.flush() return version +def _assert_update_paths_safe( + raw_json: dict[str, Any] | None, + *, + source_filename: str | None, + source_base_path: str | None, +) -> None: + if raw_json is None and source_filename is None and source_base_path is None: + return + assert_server_safe_campaign_paths( + raw_json if raw_json is not None else {}, + source_filename=source_filename, + source_base_path=source_base_path, + managed_files_available=files_integration().available, + ) + + +def _updated_runtime_json( + session: Session, + *, + tenant_id: str, + campaign: Campaign, + version: CampaignVersion, + raw_json: dict[str, Any], + source_base_path: str | None, + migrate_legacy_mail_settings: bool, +) -> dict[str, Any]: + runtime_json = normalize_campaign_paths(raw_json, source_base_path) if source_base_path else copy.deepcopy(raw_json) + requires_migration = bool(campaign_mail_profile_boundary_violations(version.raw_json)) + if requires_migration and not migrate_legacy_mail_settings: + raise CampaignPersistenceError( + "This version contains legacy campaign-local SMTP/IMAP settings. Select an authorized Mail " + "profile on the Mail settings page and explicitly save the migration; the stored legacy version " + "will not be changed automatically." + ) + assert_campaign_uses_mail_profile_reference(runtime_json) + if requires_migration and campaign_mail_profile_id(runtime_json) is None: + raise CampaignPersistenceError( + "Migrating legacy campaign mail settings requires an authorized server.mail_profile_id. " + "Select a Mail profile before saving." + ) + mail_integration().assert_campaign_mail_policy_allows_json( + session, + tenant_id=tenant_id, + raw_json=runtime_json, + campaign_id=campaign.id, + ) + return runtime_json + + +def _apply_version_field_updates( + version: CampaignVersion, + *, + current_flow: str | None, + current_step: str | None, + workflow_state: str | None, + is_complete: bool | None, + editor_state: dict[str, Any] | None, + source_filename: str | None, + source_base_path: str | None, + autosave: bool, +) -> None: + updates = ( + ("current_flow", current_flow), + ("current_step", current_step), + ("workflow_state", workflow_state), + ("is_complete", is_complete), + ("source_filename", source_filename), + ("source_base_path", source_base_path), + ) + for field_name, value in updates: + if value is not None: + setattr(version, field_name, value) + if editor_state is not None: + version.editor_state = validate_campaign_editor_state(editor_state) + if autosave: + version.autosaved_at = datetime.now(UTC) + + +def _invalidate_version_content(session: Session, *, campaign: Campaign, version: CampaignVersion) -> None: + version.validation_summary = None + version.build_summary = None + clear_execution_snapshot(version) + version.locked_at = None + version.locked_by_user_id = None + if version.workflow_state != CampaignVersionWorkflowState.EDITING.value: + version.workflow_state = CampaignVersionWorkflowState.EDITING.value + campaign.status = CampaignStatus.DRAFT.value + session.query(CampaignIssue).filter(CampaignIssue.campaign_version_id == version.id).delete(synchronize_session=False) + + +def _persist_updated_version( + session: Session, + *, + campaign: Campaign, + version: CampaignVersion, + commit: bool, +) -> None: + session.add(version) + session.add(campaign) + session.flush() + if commit: + _write_campaign_snapshot(version) + session.commit() + else: + session.flush() + + def update_campaign_version( session: Session, *, @@ -548,13 +722,7 @@ def update_campaign_version( migrate_legacy_mail_settings: bool = False, commit: bool = True, ) -> CampaignVersion: - if raw_json is not None or source_filename is not None or source_base_path is not None: - assert_server_safe_campaign_paths( - raw_json if raw_json is not None else {}, - source_filename=source_filename, - source_base_path=source_base_path, - managed_files_available=files_integration().available, - ) + _assert_update_paths_safe(raw_json, source_filename=source_filename, source_base_path=source_base_path) version = get_campaign_version_for_tenant(session, tenant_id=tenant_id, campaign_id=campaign_id, version_id=version_id) campaign = _require_campaign(session, campaign_id) ensure_current_working_version(campaign, version, action="edit") @@ -565,61 +733,35 @@ def update_campaign_version( ) if raw_json is not None: - runtime_json = normalize_campaign_paths(raw_json, source_base_path) if source_base_path else copy.deepcopy(raw_json) - if campaign_mail_profile_boundary_violations(version.raw_json) and not migrate_legacy_mail_settings: - raise CampaignPersistenceError( - "This version contains legacy campaign-local SMTP/IMAP settings. Select an authorized Mail " - "profile on the Mail settings page and explicitly save the migration; the stored legacy version " - "will not be changed automatically." - ) - assert_campaign_uses_mail_profile_reference(runtime_json) - if campaign_mail_profile_boundary_violations(version.raw_json) and campaign_mail_profile_id(runtime_json) is None: - raise CampaignPersistenceError( - "Migrating legacy campaign mail settings requires an authorized server.mail_profile_id. " - "Select a Mail profile before saving." - ) - mail_integration().assert_campaign_mail_policy_allows_json(session, tenant_id=tenant_id, raw_json=runtime_json, campaign_id=campaign.id) + runtime_json = _updated_runtime_json( + session, + tenant_id=tenant_id, + campaign=campaign, + version=version, + raw_json=raw_json, + source_base_path=source_base_path, + migrate_legacy_mail_settings=migrate_legacy_mail_settings, + ) version.raw_json = runtime_json version.schema_version = str(runtime_json.get("version", version.schema_version or "1.0")) _apply_campaign_metadata(campaign, runtime_json) - if current_flow is not None: - version.current_flow = current_flow - if current_step is not None: - version.current_step = current_step - if workflow_state is not None: - version.workflow_state = workflow_state - if is_complete is not None: - version.is_complete = is_complete - if editor_state is not None: - version.editor_state = validate_campaign_editor_state(editor_state) - if source_filename is not None: - version.source_filename = source_filename - if source_base_path is not None: - version.source_base_path = source_base_path - if autosave: - version.autosaved_at = datetime.now(UTC) + _apply_version_field_updates( + version, + current_flow=current_flow, + current_step=current_step, + workflow_state=workflow_state, + is_complete=is_complete, + editor_state=editor_state, + source_filename=source_filename, + source_base_path=source_base_path, + autosave=autosave, + ) # Changes invalidate previous build and validation summaries. if raw_json is not None: - version.validation_summary = None - version.build_summary = None - clear_execution_snapshot(version) - version.locked_at = None - version.locked_by_user_id = None - if version.workflow_state != CampaignVersionWorkflowState.EDITING.value: - version.workflow_state = CampaignVersionWorkflowState.EDITING.value - campaign.status = CampaignStatus.DRAFT.value - session.query(CampaignIssue).filter(CampaignIssue.campaign_version_id == version.id).delete(synchronize_session=False) - - session.add(version) - session.add(campaign) - session.flush() - if commit: - _write_campaign_snapshot(version) - session.commit() - else: - session.flush() + _invalidate_version_content(session, campaign=campaign, version=version) + _persist_updated_version(session, campaign=campaign, version=version, commit=commit) return version diff --git a/src/govoplan_campaign/backend/reports/campaigns.py b/src/govoplan_campaign/backend/reports/campaigns.py index 4fe8067..cdf5ab1 100644 --- a/src/govoplan_campaign/backend/reports/campaigns.py +++ b/src/govoplan_campaign/backend/reports/campaigns.py @@ -307,59 +307,40 @@ class _JobReportAggregate: include_recent_failures: bool, ) -> None: self.total += 1 + self._add_statuses(job) + self._add_delivery_counts(job, retry_max_attempts=retry_max_attempts) + self._add_issues(job) + self._add_attachments(job) + self._add_recent_failure(job, include_recent_failures=include_recent_failures) + + def _add_statuses(self, job: CampaignJob) -> None: self.build[job.build_status or "unknown"] += 1 self.validation[job.validation_status or "unknown"] += 1 self.queue[job.queue_status or "unknown"] += 1 self.send[job.send_status or "unknown"] += 1 self.postbox[job.postbox_status or "unknown"] += 1 self.imap[job.imap_status or "unknown"] += 1 + + def _add_delivery_counts(self, job: CampaignJob, *, retry_max_attempts: int | None) -> None: if job.send_status in {"queued", "claimed", "sending"}: self.pending += 1 - if ( - job.validation_status in {"ready", "warning"} - and job.build_status == "built" - ): + if _job_is_queueable(job): self.queueable += 1 - if ( - job.attempt_count == 0 - and job.postbox_attempt_count == 0 - and job.send_status in {"not_queued", "cancelled"} - and job.validation_status in {"ready", "warning"} - and job.build_status == "built" - ): + if _job_is_queueable_unattempted(job): self.queueable_unattempted += 1 - if ( - job.send_status in {"failed_temporary", "partially_accepted"} - and ( - retry_max_attempts is None - or job.attempt_count < retry_max_attempts - ) - ): + if _job_is_retryable(job, retry_max_attempts=retry_max_attempts): self.retryable += 1 - if job.send_status not in { - "skipped", - "smtp_accepted", - "postbox_accepted", - "delivered", - "partially_accepted", - "sent", - "outcome_unknown", - "claimed", - "sending", - "cancelled", - }: + if _job_is_cancellable(job): self.cancellable += 1 if _job_needs_attention(job): self.needs_attention += 1 - self._add_issues(job) - self._add_attachments(job) - if include_recent_failures and _job_is_recent_failure(job): - self.recent_failures.append(job) - self.recent_failures.sort( - key=lambda item: item.updated_at or item.created_at, - reverse=True, - ) - del self.recent_failures[20:] + + def _add_recent_failure(self, job: CampaignJob, *, include_recent_failures: bool) -> None: + if not include_recent_failures or not _job_is_recent_failure(job): + return + self.recent_failures.append(job) + self.recent_failures.sort(key=lambda item: item.updated_at or item.created_at, reverse=True) + del self.recent_failures[20:] def _add_issues(self, job: CampaignJob) -> None: for issue in job.issues_snapshot or []: @@ -391,6 +372,42 @@ class _JobReportAggregate: self.attachment_ambiguous += 1 +NON_CANCELLABLE_SEND_STATUSES = { + "skipped", + "smtp_accepted", + "postbox_accepted", + "delivered", + "partially_accepted", + "sent", + "outcome_unknown", + "claimed", + "sending", + "cancelled", +} + + +def _job_is_queueable(job: CampaignJob) -> bool: + return job.validation_status in {"ready", "warning"} and job.build_status == "built" + + +def _job_is_queueable_unattempted(job: CampaignJob) -> bool: + return ( + job.attempt_count == 0 + and job.postbox_attempt_count == 0 + and job.send_status in {"not_queued", "cancelled"} + and _job_is_queueable(job) + ) + + +def _job_is_retryable(job: CampaignJob, *, retry_max_attempts: int | None) -> bool: + attempts_available = retry_max_attempts is None or job.attempt_count < retry_max_attempts + return job.send_status in {"failed_temporary", "partially_accepted"} and attempts_available + + +def _job_is_cancellable(job: CampaignJob) -> bool: + return job.send_status not in NON_CANCELLABLE_SEND_STATUSES + + def _job_needs_attention(job: CampaignJob) -> bool: return ( job.validation_status in {"needs_review", "blocked"} @@ -577,31 +594,48 @@ def _job_evidence_row( "campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id, "message_id_header": job.message_id_header, - "from": _address_summary(recipients.get("from")), - "to": _address_summary(recipients.get("to")), - "cc": _address_summary(recipients.get("cc")), - "bcc": _address_summary(recipients.get("bcc")), - "reply_to": _address_summary(recipients.get("reply_to")), - "postbox_targets": "; ".join( - str(target.get("address") or target.get("name") or target.get("postbox_id") or "") - for target in ( - getattr(job, "resolved_postbox_targets", None) or [] - ) - if isinstance(target, dict) - ), + **_job_evidence_addresses(recipients), + "postbox_targets": _job_postbox_target_summary(job), "attachment_names": _attachment_names(job.resolved_attachments), + **_job_attempt_evidence(latest_smtp=latest_smtp, latest_imap=latest_imap), + }) + return row + + +def _job_evidence_addresses(recipients: dict[str, Any]) -> dict[str, str]: + return {key: _address_summary(recipients.get(key)) for key in ("from", "to", "cc", "bcc", "reply_to")} + + +def _job_postbox_target_summary(job: CampaignJob) -> str: + targets = getattr(job, "resolved_postbox_targets", None) or [] + return "; ".join( + str(target.get("address") or target.get("name") or target.get("postbox_id") or "") + for target in targets + if isinstance(target, dict) + ) + + +def _iso_timestamp(value: datetime | None) -> str | None: + return value.isoformat() if value else None + + +def _job_attempt_evidence( + *, + latest_smtp: SendAttempt | None, + latest_imap: ImapAppendAttempt | None, +) -> dict[str, Any]: + return { "latest_smtp_attempt_number": latest_smtp.attempt_number if latest_smtp else None, "latest_smtp_status": latest_smtp.status if latest_smtp else None, "latest_smtp_status_code": latest_smtp.smtp_status_code if latest_smtp else None, - "latest_smtp_started_at": latest_smtp.started_at.isoformat() if latest_smtp and latest_smtp.started_at else None, - "latest_smtp_finished_at": latest_smtp.finished_at.isoformat() if latest_smtp and latest_smtp.finished_at else None, + "latest_smtp_started_at": _iso_timestamp(latest_smtp.started_at) if latest_smtp else None, + "latest_smtp_finished_at": _iso_timestamp(latest_smtp.finished_at) if latest_smtp else None, "latest_imap_attempt_number": latest_imap.attempt_number if latest_imap else None, "latest_imap_status": latest_imap.status if latest_imap else None, "latest_imap_folder": latest_imap.folder if latest_imap else None, - "latest_imap_created_at": latest_imap.created_at.isoformat() if latest_imap and latest_imap.created_at else None, - "latest_imap_updated_at": latest_imap.updated_at.isoformat() if latest_imap and latest_imap.updated_at else None, - }) - return row + "latest_imap_created_at": _iso_timestamp(latest_imap.created_at) if latest_imap else None, + "latest_imap_updated_at": _iso_timestamp(latest_imap.updated_at) if latest_imap else None, + } def generate_campaign_report( diff --git a/src/govoplan_campaign/backend/router.py b/src/govoplan_campaign/backend/router.py index 932e9f3..37d2622 100644 --- a/src/govoplan_campaign/backend/router.py +++ b/src/govoplan_campaign/backend/router.py @@ -2712,8 +2712,7 @@ def _campaign_jobs_page_response( cursor: str | None = None, changed_job_ids: set[str] | None = None, ) -> CampaignJobsResponse: - total_unfiltered = int(session.query(func.count(CampaignJob.id)).filter(*base_filters).scalar() or 0) - total = int(session.query(func.count(CampaignJob.id)).filter(*filtered).scalar() or 0) + total_unfiltered, total = _campaign_jobs_page_counts(session, base_filters=base_filters, filtered=filtered) pages = (total + page_size - 1) // page_size if total else 0 fingerprint = _campaign_jobs_cursor_fingerprint( campaign_id=campaign_id, @@ -2728,47 +2727,26 @@ def _campaign_jobs_page_response( sort_direction=sort_direction, ) ordering = _campaign_jobs_ordering(sort_by, sort_direction) - ordered_query = session.query(CampaignJob).filter(*filtered).order_by(*ordering) - start_cursor: str | None = None - effective_offset = 0 - if cursor: - if sort_by != "number" or sort_direction != "asc": - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Campaign job cursors require number ascending order", - ) - try: - cursor_values = decode_keyset_cursor(CAMPAIGN_JOBS_CURSOR_SCOPE, cursor, fingerprint=fingerprint) - if cursor_values is None: - raise KeysetCursorError("Invalid pagination cursor") - page_query = session.query(CampaignJob).filter(*filtered).filter(_campaign_jobs_cursor_condition(cursor_values)) - except KeysetCursorError as exc: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - start_cursor = cursor - else: - effective_offset = (page - 1) * page_size - page_query = session.query(CampaignJob).filter(*filtered) - if effective_offset > 0 and sort_by == "number" and sort_direction == "asc": - previous_row = ordered_query.offset(effective_offset - 1).limit(1).first() - if previous_row is not None: - start_cursor = _campaign_jobs_cursor_for_row(previous_row, fingerprint=fingerprint) - - rows_plus_one = ( - page_query - .order_by(*ordering) - .offset(effective_offset) - .limit(page_size + 1) - .all() + rows_plus_one, start_cursor = _campaign_jobs_page_rows( + session, + filtered=filtered, + ordering=ordering, + page=page, + page_size=page_size, + sort_by=sort_by, + sort_direction=sort_direction, + cursor=cursor, + fingerprint=fingerprint, ) jobs = rows_plus_one[:page_size] - next_cursor = ( - _campaign_jobs_cursor_for_row(jobs[-1], fingerprint=fingerprint) - if changed_job_ids is None - and sort_by == "number" - and sort_direction == "asc" - and len(rows_plus_one) > page_size - and jobs - else None + next_cursor = _campaign_jobs_next_cursor( + jobs, + rows_plus_one=rows_plus_one, + page_size=page_size, + sort_by=sort_by, + sort_direction=sort_direction, + changed_job_ids=changed_job_ids, + fingerprint=fingerprint, ) if changed_job_ids is not None: jobs = [job for job in jobs if job.id in changed_job_ids] @@ -2787,6 +2765,110 @@ def _campaign_jobs_page_response( ) +def _campaign_jobs_page_counts( + session: Session, + *, + base_filters: list[object], + filtered: list[object], +) -> tuple[int, int]: + total_unfiltered = int(session.query(func.count(CampaignJob.id)).filter(*base_filters).scalar() or 0) + total = int(session.query(func.count(CampaignJob.id)).filter(*filtered).scalar() or 0) + return total_unfiltered, total + + +def _campaign_jobs_page_rows( + session: Session, + *, + filtered: list[object], + ordering: list[object], + page: int, + page_size: int, + sort_by: str, + sort_direction: str, + cursor: str | None, + fingerprint: str, +) -> tuple[list[CampaignJob], str | None]: + if cursor: + page_query = _campaign_jobs_query_after_cursor( + session, + filtered=filtered, + cursor=cursor, + fingerprint=fingerprint, + sort_by=sort_by, + sort_direction=sort_direction, + ) + return page_query.order_by(*ordering).limit(page_size + 1).all(), cursor + + effective_offset = (page - 1) * page_size + page_query = session.query(CampaignJob).filter(*filtered) + start_cursor = _campaign_jobs_offset_cursor( + page_query, + ordering=ordering, + effective_offset=effective_offset, + sort_by=sort_by, + sort_direction=sort_direction, + fingerprint=fingerprint, + ) + rows = page_query.order_by(*ordering).offset(effective_offset).limit(page_size + 1).all() + return rows, start_cursor + + +def _campaign_jobs_query_after_cursor( + session: Session, + *, + filtered: list[object], + cursor: str, + fingerprint: str, + sort_by: str, + sort_direction: str, +): + if sort_by != "number" or sort_direction != "asc": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Campaign job cursors require number ascending order", + ) + try: + cursor_values = decode_keyset_cursor(CAMPAIGN_JOBS_CURSOR_SCOPE, cursor, fingerprint=fingerprint) + if cursor_values is None: + raise KeysetCursorError("Invalid pagination cursor") + except KeysetCursorError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + return session.query(CampaignJob).filter(*filtered).filter(_campaign_jobs_cursor_condition(cursor_values)) + + +def _campaign_jobs_offset_cursor( + query, + *, + ordering: list[object], + effective_offset: int, + sort_by: str, + sort_direction: str, + fingerprint: str, +) -> str | None: + if effective_offset <= 0 or sort_by != "number" or sort_direction != "asc": + return None + previous_row = query.order_by(*ordering).offset(effective_offset - 1).limit(1).first() + if previous_row is None: + return None + return _campaign_jobs_cursor_for_row(previous_row, fingerprint=fingerprint) + + +def _campaign_jobs_next_cursor( + jobs: list[CampaignJob], + *, + rows_plus_one: list[CampaignJob], + page_size: int, + sort_by: str, + sort_direction: str, + changed_job_ids: set[str] | None, + fingerprint: str, +) -> str | None: + cursor_supported = sort_by == "number" and sort_direction == "asc" + if changed_job_ids is not None or not cursor_supported or len(rows_plus_one) <= page_size or not jobs: + return None + return _campaign_jobs_cursor_for_row(jobs[-1], fingerprint=fingerprint) + + def _campaign_jobs_cursor_fingerprint( *, campaign_id: str, diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index 4098a60..2c0b14a 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -1,9 +1,9 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Literal +from typing import Annotated, Any, Literal -from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator +from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, ValidationInfo, field_validator, model_validator from govoplan_core.api.v1.schemas import DeltaDeletedItem from govoplan_campaign.backend.campaign.mail_profile_boundary import ( @@ -570,62 +570,67 @@ class CampaignActionResponse(BaseModel): result: dict[str, Any] +def _valid_report_email_domain(domain: str) -> bool: + if not domain or domain.startswith(".") or domain.endswith(".") or ".." in domain: + return False + return all( + label + and not label.startswith("-") + and not label.endswith("-") + and all(character.isalnum() or character == "-" for character in label) + for label in domain.split(".") + ) + + +def _normalize_report_recipient(value: Any) -> str: + if not isinstance(value, str): + raise ValueError("report recipients must be email-address strings") + recipient = value.strip() + if len(recipient) > 320: + raise ValueError("report recipient addresses must be at most 320 characters") + if any(ord(character) < 32 or ord(character) == 127 for character in recipient): + raise ValueError("report recipient addresses must not contain control characters") + if recipient.count("@") != 1: + raise ValueError("report recipients must be email addresses") + local, domain = recipient.split("@", 1) + invalid_local = not local or local.startswith(".") or local.endswith(".") or ".." in local + invalid_address = ( + any(character.isspace() for character in recipient) + or any(character in ',;:<>[]()\\"' for character in recipient) + ) + if invalid_local or invalid_address or not _valid_report_email_domain(domain): + raise ValueError("report recipients must be email addresses") + return recipient + + +ReportEmailAddress = Annotated[str, BeforeValidator(_normalize_report_recipient)] + + +def _deduplicate_report_recipients(value: list[str]) -> list[str]: + recipients: list[str] = [] + seen: set[str] = set() + for recipient in value: + key = recipient.casefold() + if key not in seen: + seen.add(key) + recipients.append(recipient) + return recipients + + class ReportEmailRequest(BaseModel): model_config = ConfigDict(extra="forbid") - to: list[str] = Field(min_length=1, max_length=50) + to: list[ReportEmailAddress] = Field(min_length=1, max_length=50) version_id: str | None = None include_jobs: bool = False attach_jobs_csv: bool = False attach_report_json: bool = False dry_run: bool = False - @field_validator("to", mode="before") + @field_validator("to") @classmethod - def normalize_and_validate_recipients(cls, value: Any) -> Any: - if not isinstance(value, list): - return value - if not 1 <= len(value) <= 50: - raise ValueError("report email requires between 1 and 50 recipients") - recipients: list[str] = [] - seen: set[str] = set() - for item in value: - if not isinstance(item, str): - raise ValueError("report recipients must be email-address strings") - recipient = item.strip() - if len(recipient) > 320: - raise ValueError("report recipient addresses must be at most 320 characters") - if any(ord(character) < 32 or ord(character) == 127 for character in recipient): - raise ValueError("report recipient addresses must not contain control characters") - if recipient.count("@") != 1: - raise ValueError("report recipients must be email addresses") - local, domain = recipient.split("@", 1) - if ( - not local - or not domain - or any(character.isspace() for character in recipient) - or any(character in ',;:<>[]()\\"' for character in recipient) - or local.startswith(".") - or local.endswith(".") - or ".." in local - or domain.startswith(".") - or domain.endswith(".") - or ".." in domain - or any( - not label - or label.startswith("-") - or label.endswith("-") - or not all(character.isalnum() or character == "-" for character in label) - for label in domain.split(".") - ) - ): - raise ValueError("report recipients must be email addresses") - key = recipient.casefold() - if key in seen: - continue - seen.add(key) - recipients.append(recipient) - return recipients + def normalize_and_validate_recipients(cls, value: list[str]) -> list[str]: + return _deduplicate_report_recipients(value) class ReportEmailResponse(BaseModel): diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index 2f376e1..04f5155 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -222,6 +222,33 @@ class _MailChannelOutcome: ) +@dataclass(frozen=True, slots=True) +class _DeliveryOutcomeSummary: + mail_accepted: bool + postbox_accepted: int + postbox_rejected: int + outcome_unknown: bool + temporary_rejection: bool + + @property + def accepted_count(self) -> int: + return int(self.mail_accepted) + self.postbox_accepted + + +@dataclass(frozen=True, slots=True) +class _ImapAppendContext: + snapshot: ExecutionSnapshot + message_bytes: bytes + folder: str + + +@dataclass(frozen=True, slots=True) +class _ClaimedImapAppend: + job: CampaignJob + attempt: ImapAppendAttempt + claim_token: str + + QUEUEABLE_VALIDATION_STATUSES = { JobValidationStatus.READY.value, JobValidationStatus.WARNING.value, @@ -2338,47 +2365,50 @@ def _final_multichannel_status( mail: _MailChannelOutcome | None, postbox: PostboxChannelOutcome | None, ) -> str: - mail_accepted = bool(mail and mail.accepted) - postbox_accepted = int(postbox.accepted_count if postbox else 0) - postbox_rejected = int(postbox.rejected_count if postbox else 0) - unknown = bool( - (mail and mail.outcome_unknown) - or (postbox and postbox.outcome_unknown) - ) - if unknown: + outcome = _delivery_outcome_summary(mail=mail, postbox=postbox) + if outcome.outcome_unknown: return JobSendStatus.OUTCOME_UNKNOWN.value + if not outcome.accepted_count: + return JobSendStatus.FAILED_TEMPORARY.value if outcome.temporary_rejection else JobSendStatus.FAILED_PERMANENT.value + classifier = _ACCEPTED_DELIVERY_CLASSIFIERS.get(channel_policy, _classify_fallback_delivery) + return classifier(outcome) - accepted_count = int(mail_accepted) + postbox_accepted - if accepted_count: - if channel_policy == DeliveryChannelPolicy.POSTBOX: - return ( - JobSendStatus.PARTIALLY_ACCEPTED.value - if postbox_rejected - else JobSendStatus.POSTBOX_ACCEPTED.value - ) - if channel_policy == DeliveryChannelPolicy.MAIL_AND_POSTBOX: - if mail_accepted and postbox_accepted and not postbox_rejected: - return JobSendStatus.DELIVERED.value - return JobSendStatus.PARTIALLY_ACCEPTED.value - if postbox_rejected: - return JobSendStatus.PARTIALLY_ACCEPTED.value - return ( - JobSendStatus.SMTP_ACCEPTED.value - if mail_accepted - else JobSendStatus.POSTBOX_ACCEPTED.value - ) - temporary = bool( - (mail and mail.rejected_temporary) - or (postbox and postbox.rejected_temporary) - ) - return ( - JobSendStatus.FAILED_TEMPORARY.value - if temporary - else JobSendStatus.FAILED_PERMANENT.value +def _delivery_outcome_summary( + *, + mail: _MailChannelOutcome | None, + postbox: PostboxChannelOutcome | None, +) -> _DeliveryOutcomeSummary: + return _DeliveryOutcomeSummary( + mail_accepted=bool(mail and mail.accepted), + postbox_accepted=int(postbox.accepted_count if postbox else 0), + postbox_rejected=int(postbox.rejected_count if postbox else 0), + outcome_unknown=bool((mail and mail.outcome_unknown) or (postbox and postbox.outcome_unknown)), + temporary_rejection=bool((mail and mail.rejected_temporary) or (postbox and postbox.rejected_temporary)), ) +def _classify_postbox_delivery(outcome: _DeliveryOutcomeSummary) -> str: + return JobSendStatus.PARTIALLY_ACCEPTED.value if outcome.postbox_rejected else JobSendStatus.POSTBOX_ACCEPTED.value + + +def _classify_dual_delivery(outcome: _DeliveryOutcomeSummary) -> str: + fully_delivered = outcome.mail_accepted and outcome.postbox_accepted > 0 and not outcome.postbox_rejected + return JobSendStatus.DELIVERED.value if fully_delivered else JobSendStatus.PARTIALLY_ACCEPTED.value + + +def _classify_fallback_delivery(outcome: _DeliveryOutcomeSummary) -> str: + if outcome.postbox_rejected: + return JobSendStatus.PARTIALLY_ACCEPTED.value + return JobSendStatus.SMTP_ACCEPTED.value if outcome.mail_accepted else JobSendStatus.POSTBOX_ACCEPTED.value + + +_ACCEPTED_DELIVERY_CLASSIFIERS = { + DeliveryChannelPolicy.POSTBOX: _classify_postbox_delivery, + DeliveryChannelPolicy.MAIL_AND_POSTBOX: _classify_dual_delivery, +} + + def _multichannel_messages( mail: _MailChannelOutcome | None, postbox: PostboxChannelOutcome | None, @@ -2931,22 +2961,29 @@ def _mark_imap_append_outcome_unknown_after_effect( ) -def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) -> AppendSentResult: - """Append one successfully sent job's exact EML to the configured IMAP Sent folder.""" +def _imap_append_precondition( + session: Session, + job: CampaignJob, + *, + dry_run: bool, +) -> AppendSentResult | None: + if not _mail_was_accepted(session, job.id, send_status=job.send_status): + return AppendSentResult( + job_id=job.id, + status="not_sent", + attempt_number=0, + dry_run=dry_run, + message="SMTP has not accepted this job", + ) + return _imap_blocked_result(session, job, dry_run=dry_run) - job = session.get(CampaignJob, job_id) - if not job: - raise SendJobError(f"Job not found: {job_id}") - if not _mail_was_accepted( - session, - job.id, - send_status=job.send_status, - ): - return AppendSentResult(job_id=job_id, status="not_sent", attempt_number=0, dry_run=dry_run, message="SMTP has not accepted this job") - blocked = _imap_blocked_result(session, job, dry_run=dry_run) - if blocked is not None: - return blocked +def _prepare_imap_append( + session: Session, + job: CampaignJob, + *, + dry_run: bool, +) -> _ImapAppendContext | AppendSentResult: version = session.get(CampaignVersion, job.campaign_version_id) if not version: raise SendJobError("Campaign version not found") @@ -2965,20 +3002,25 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) session.add(job) session.commit() return AppendSentResult(job_id=job.id, status="skipped", attempt_number=0, dry_run=dry_run, message=job.last_error) + return _ImapAppendContext( + snapshot=snapshot, + message_bytes=_load_eml_bytes_for_job(job), + folder=snapshot.delivery.imap_append_sent.folder or "auto", + ) - message_bytes = _load_eml_bytes_for_job(job) - folder = snapshot.delivery.imap_append_sent.folder or "auto" - if dry_run: - attempts = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id == job.id).count() - return AppendSentResult( - job_id=job.id, - status="dry_run", - attempt_number=attempts, - dry_run=True, - folder=folder, - message=f"Would append {len(message_bytes)} bytes to IMAP folder {folder!r}", - ) +def _imap_append_dry_run(session: Session, job: CampaignJob, context: _ImapAppendContext) -> AppendSentResult: + return AppendSentResult( + job_id=job.id, + status="dry_run", + attempt_number=_imap_attempt_count(session, job.id), + dry_run=True, + folder=context.folder, + message=f"Would append {len(context.message_bytes)} bytes to IMAP folder {context.folder!r}", + ) + + +def _claim_imap_append(session: Session, job: CampaignJob) -> _ClaimedImapAppend | AppendSentResult: claim_token = _claim_job_for_imap_append(session, job) if claim_token is None: current = session.get(CampaignJob, job.id) @@ -2993,80 +3035,98 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) attempt_number=_imap_attempt_count(session, current.id), message=f"Job is no longer eligible for IMAP append: {current.imap_status}", ) - job = session.get(CampaignJob, job.id) - if job is None: + claimed_job = session.get(CampaignJob, job.id) + if claimed_job is None: raise SendJobError("Claimed campaign job disappeared before IMAP append") - attempt = _record_imap_attempt_start(session, job, claim_token) + return _ClaimedImapAppend( + job=claimed_job, + attempt=_record_imap_attempt_start(session, claimed_job, claim_token), + claim_token=claim_token, + ) + + +def _record_imap_provider_error( + session: Session, + claimed: _ClaimedImapAppend, + *, + folder: str, + message: str, + outcome_unknown: bool, +) -> None: + owned = _record_imap_append_failure( + session, + job=claimed.job, + attempt=claimed.attempt, + claim_token=claimed.claim_token, + folder=folder, + message=message, + outcome_unknown=outcome_unknown, + ) + if not owned: + _imap_result_after_lost_claim( + session, + job_id=claimed.job.id, + attempt=claimed.attempt, + provider_succeeded=outcome_unknown, + ) + + +def _invoke_imap_append(session: Session, claimed: _ClaimedImapAppend, context: _ImapAppendContext): + snapshot = context.snapshot + job = claimed.job + return mail_integration().append_campaign_message_to_sent( + session, + tenant_id=job.tenant_id, + campaign_id=job.campaign_id, + profile_id=snapshot.mail_profile_id, + message_bytes=context.message_bytes, + folder=None if context.folder == "auto" else context.folder, + expected_smtp_transport_revision=snapshot.smtp_transport_revision or "", + expected_imap_transport_revision=snapshot.imap_transport_revision, + smtp_server_id=snapshot.smtp_server_id, + smtp_credential_id=snapshot.smtp_credential_id, + imap_server_id=snapshot.imap_server_id, + imap_credential_id=snapshot.imap_credential_id, + ) + + +def _perform_imap_append( + session: Session, + claimed: _ClaimedImapAppend, + context: _ImapAppendContext, +) -> AppendSentResult: try: - result = mail_integration().append_campaign_message_to_sent( - session, - tenant_id=job.tenant_id, - campaign_id=job.campaign_id, - profile_id=snapshot.mail_profile_id, - message_bytes=message_bytes, - folder=None if folder == "auto" else folder, - expected_smtp_transport_revision=snapshot.smtp_transport_revision or "", - expected_imap_transport_revision=snapshot.imap_transport_revision, - smtp_server_id=snapshot.smtp_server_id, - smtp_credential_id=snapshot.smtp_credential_id, - imap_server_id=snapshot.imap_server_id, - imap_credential_id=snapshot.imap_credential_id, - ) + result = _invoke_imap_append(session, claimed, context) except (MailProfileError, ImapConfigurationError, ImapAppendError) as exc: - outcome_unknown = bool(getattr(exc, "outcome_unknown", False)) - owned = _record_imap_append_failure( + _record_imap_provider_error( session, - job=job, - attempt=attempt, - claim_token=claim_token, - folder=folder, + claimed, + folder=context.folder, message=str(exc), - outcome_unknown=outcome_unknown, + outcome_unknown=bool(getattr(exc, "outcome_unknown", False)), ) - if not owned: - _imap_result_after_lost_claim( - session, - job_id=job.id, - attempt=attempt, - provider_succeeded=outcome_unknown, - ) raise except Exception: reason = ( "The Sent-folder append outcome is unknown after an unexpected provider failure; " "inspect and reconcile the mailbox before retrying." ) - owned = _record_imap_append_failure( - session, - job=job, - attempt=attempt, - claim_token=claim_token, - folder=folder, - message=reason, - outcome_unknown=True, - ) - if not owned: - _imap_result_after_lost_claim( - session, - job_id=job.id, - attempt=attempt, - provider_succeeded=True, - ) + _record_imap_provider_error(session, claimed, folder=context.folder, message=reason, outcome_unknown=True) raise ImapAppendError(reason, outcome_unknown=True) from None try: return _record_imap_append_success( session, - job=job, - attempt=attempt, - claim_token=claim_token, + job=claimed.job, + attempt=claimed.attempt, + claim_token=claimed.claim_token, folder=result.folder, ) except Exception: return _mark_imap_append_outcome_unknown_after_effect( session, - job_id=job.id, - attempt_id=attempt.id, - claim_token=claim_token, + job_id=claimed.job.id, + attempt_id=claimed.attempt.id, + claim_token=claimed.claim_token, reason=( "The IMAP provider accepted the append, but persisting the outcome failed. " "Automatic retry is stopped until an operator reconciles the mailbox." @@ -3074,6 +3134,26 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) ) +def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) -> AppendSentResult: + """Append one successfully sent job's exact EML to the configured IMAP Sent folder.""" + + job = session.get(CampaignJob, job_id) + if not job: + raise SendJobError(f"Job not found: {job_id}") + blocked = _imap_append_precondition(session, job, dry_run=dry_run) + if blocked is not None: + return blocked + prepared = _prepare_imap_append(session, job, dry_run=dry_run) + if isinstance(prepared, AppendSentResult): + return prepared + if dry_run: + return _imap_append_dry_run(session, job, prepared) + claimed = _claim_imap_append(session, job) + if isinstance(claimed, AppendSentResult): + return claimed + return _perform_imap_append(session, claimed, prepared) + + def enqueue_pending_imap_appends( session: Session, *, diff --git a/src/govoplan_campaign/backend/sending/postbox_delivery.py b/src/govoplan_campaign/backend/sending/postbox_delivery.py index 953c842..d049023 100644 --- a/src/govoplan_campaign/backend/sending/postbox_delivery.py +++ b/src/govoplan_campaign/backend/sending/postbox_delivery.py @@ -173,97 +173,82 @@ def _participants(job: CampaignJob) -> tuple[PostboxParticipantRef, ...]: def _attachments(job: CampaignJob) -> tuple[PostboxAttachmentRef, ...]: - values = [ - PostboxAttachmentRef( - reference_type="campaign_eml", - reference_id=job.id, - name=f"{job.entry_id or job.entry_index}.eml", - media_type="message/rfc822", - size_bytes=job.eml_size_bytes, - digest=job.eml_sha256, - metadata={ - "campaign_id": job.campaign_id, - "campaign_version_id": job.campaign_version_id, - }, - ) - ] + values = [_eml_attachment(job)] for rule_index, rule in enumerate(job.resolved_attachments or []): if not isinstance(rule, dict): continue managed_matches = rule.get("managed_matches") if isinstance(managed_matches, list) and managed_matches: - for match in managed_matches: - if not isinstance(match, dict): - continue - reference_id = str( - match.get("version_id") - or match.get("asset_id") - or match.get("blob_id") - or "" - ).strip() - if not reference_id: - continue - values.append( - PostboxAttachmentRef( - reference_type=( - "file_version" - if match.get("version_id") - else "file_asset" - ), - reference_id=reference_id, - name=str(match.get("filename") or "").strip() or None, - media_type=( - str(match.get("content_type")) - if match.get("content_type") - else None - ), - size_bytes=( - int(match["size_bytes"]) - if match.get("size_bytes") is not None - else None - ), - digest=( - str(match.get("checksum_sha256")) - if match.get("checksum_sha256") - else None - ), - metadata={ - key: value - for key, value in match.items() - if key - in { - "asset_id", - "version_id", - "blob_id", - "display_path", - "relative_path", - "owner_type", - "owner_id", - "source_revision", - } - }, - ) - ) + values.extend(_managed_attachments(managed_matches)) continue matches = rule.get("matches") - if not isinstance(matches, list): - continue - for match_index, match in enumerate(matches): - values.append( - PostboxAttachmentRef( - reference_type="campaign_attachment", - reference_id=f"{job.id}:{rule_index}:{match_index}", - name=str(match).rsplit("/", 1)[-1] or None, - metadata={ - "campaign_id": job.campaign_id, - "campaign_version_id": job.campaign_version_id, - "job_id": job.id, - }, - ) - ) + if isinstance(matches, list): + values.extend(_campaign_attachments(job, rule_index=rule_index, matches=matches)) return tuple(values) +def _eml_attachment(job: CampaignJob) -> PostboxAttachmentRef: + return PostboxAttachmentRef( + reference_type="campaign_eml", + reference_id=job.id, + name=f"{job.entry_id or job.entry_index}.eml", + media_type="message/rfc822", + size_bytes=job.eml_size_bytes, + digest=job.eml_sha256, + metadata={"campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id}, + ) + + +MANAGED_ATTACHMENT_METADATA_KEYS = { + "asset_id", + "version_id", + "blob_id", + "display_path", + "relative_path", + "owner_type", + "owner_id", + "source_revision", +} + + +def _managed_attachment(match: dict[str, Any]) -> PostboxAttachmentRef | None: + reference_id = str(match.get("version_id") or match.get("asset_id") or match.get("blob_id") or "").strip() + if not reference_id: + return None + return PostboxAttachmentRef( + reference_type="file_version" if match.get("version_id") else "file_asset", + reference_id=reference_id, + name=str(match.get("filename") or "").strip() or None, + media_type=str(match.get("content_type")) if match.get("content_type") else None, + size_bytes=int(match["size_bytes"]) if match.get("size_bytes") is not None else None, + digest=str(match.get("checksum_sha256")) if match.get("checksum_sha256") else None, + metadata={key: value for key, value in match.items() if key in MANAGED_ATTACHMENT_METADATA_KEYS}, + ) + + +def _managed_attachments(matches: list[Any]) -> list[PostboxAttachmentRef]: + attachments = (_managed_attachment(match) for match in matches if isinstance(match, dict)) + return [attachment for attachment in attachments if attachment is not None] + + +def _campaign_attachments( + job: CampaignJob, + *, + rule_index: int, + matches: list[Any], +) -> list[PostboxAttachmentRef]: + metadata = {"campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id, "job_id": job.id} + return [ + PostboxAttachmentRef( + reference_type="campaign_attachment", + reference_id=f"{job.id}:{rule_index}:{match_index}", + name=str(match).rsplit("/", 1)[-1] or None, + metadata=metadata, + ) + for match_index, match in enumerate(matches) + ] + + def _sender_label(job: CampaignJob) -> str | None: recipients = job.resolved_recipients or {} sender = recipients.get("from") diff --git a/tests/test_postbox_delivery_orchestration.py b/tests/test_postbox_delivery_orchestration.py index c559d42..955a0a8 100644 --- a/tests/test_postbox_delivery_orchestration.py +++ b/tests/test_postbox_delivery_orchestration.py @@ -4,6 +4,8 @@ import unittest from types import SimpleNamespace from unittest.mock import patch +import pytest + from govoplan_campaign.backend.campaign.models import DeliveryChannelPolicy from govoplan_campaign.backend.db.models import JobSendStatus from govoplan_campaign.backend.sending.jobs import ( @@ -285,5 +287,114 @@ class PostboxFallbackOrchestrationTests(unittest.TestCase): ) +@pytest.mark.parametrize("policy", list(DeliveryChannelPolicy)) +@pytest.mark.parametrize( + ("mail", "postbox"), + [ + (_MailChannelOutcome(outcome_unknown=True), PostboxChannelOutcome()), + (_MailChannelOutcome(accepted=True), PostboxChannelOutcome(outcome_unknown=1)), + (_MailChannelOutcome(outcome_unknown=True), PostboxChannelOutcome(accepted=1)), + ], +) +def test_outcome_unknown_has_precedence_for_every_delivery_policy( + policy: DeliveryChannelPolicy, + mail: _MailChannelOutcome, + postbox: PostboxChannelOutcome, +) -> None: + assert _final_multichannel_status(channel_policy=policy, mail=mail, postbox=postbox) == JobSendStatus.OUTCOME_UNKNOWN.value + + +@pytest.mark.parametrize("policy", list(DeliveryChannelPolicy)) +@pytest.mark.parametrize( + ("mail", "postbox", "expected"), + [ + ( + _MailChannelOutcome(rejected_permanent=True), + PostboxChannelOutcome(rejected_permanent=1), + JobSendStatus.FAILED_PERMANENT.value, + ), + ( + _MailChannelOutcome(rejected_temporary=True), + PostboxChannelOutcome(rejected_permanent=1), + JobSendStatus.FAILED_TEMPORARY.value, + ), + ( + _MailChannelOutcome(rejected_permanent=True), + PostboxChannelOutcome(rejected_temporary=1), + JobSendStatus.FAILED_TEMPORARY.value, + ), + ], +) +def test_rejection_precedence_is_exhaustive_for_every_delivery_policy( + policy: DeliveryChannelPolicy, + mail: _MailChannelOutcome, + postbox: PostboxChannelOutcome, + expected: str, +) -> None: + assert _final_multichannel_status(channel_policy=policy, mail=mail, postbox=postbox) == expected + + +@pytest.mark.parametrize( + ("policy", "mail", "postbox", "expected"), + [ + ( + DeliveryChannelPolicy.MAIL, + _MailChannelOutcome(accepted=True), + PostboxChannelOutcome(), + JobSendStatus.SMTP_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.POSTBOX, + None, + PostboxChannelOutcome(accepted=1), + JobSendStatus.POSTBOX_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.POSTBOX, + None, + PostboxChannelOutcome(accepted=1, rejected_permanent=1), + JobSendStatus.PARTIALLY_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.MAIL_AND_POSTBOX, + _MailChannelOutcome(accepted=True), + PostboxChannelOutcome(accepted=1), + JobSendStatus.DELIVERED.value, + ), + ( + DeliveryChannelPolicy.MAIL_AND_POSTBOX, + _MailChannelOutcome(accepted=True), + PostboxChannelOutcome(rejected_permanent=1), + JobSendStatus.PARTIALLY_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.MAIL_AND_POSTBOX, + _MailChannelOutcome(rejected_permanent=True), + PostboxChannelOutcome(accepted=1), + JobSendStatus.PARTIALLY_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.MAIL_THEN_POSTBOX, + _MailChannelOutcome(rejected_permanent=True), + PostboxChannelOutcome(accepted=1), + JobSendStatus.POSTBOX_ACCEPTED.value, + ), + ( + DeliveryChannelPolicy.POSTBOX_THEN_MAIL, + _MailChannelOutcome(accepted=True), + PostboxChannelOutcome(rejected_permanent=1), + JobSendStatus.PARTIALLY_ACCEPTED.value, + ), + ], +) +def test_accepted_delivery_decision_table( + policy: DeliveryChannelPolicy, + mail: _MailChannelOutcome | None, + postbox: PostboxChannelOutcome, + expected: str, +) -> None: + assert _final_multichannel_status(channel_policy=policy, mail=mail, postbox=postbox) == expected + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_report_email_security.py b/tests/test_report_email_security.py index 141bf0d..c87ba34 100644 --- a/tests/test_report_email_security.py +++ b/tests/test_report_email_security.py @@ -42,6 +42,16 @@ def test_report_email_recipient_schema_rejects_unsafe_addresses(recipients: list ReportEmailRequest.model_validate({"to": recipients}) +def test_report_email_recipient_error_identifies_the_invalid_list_item() -> None: + with pytest.raises(ValidationError) as captured: + ReportEmailRequest.model_validate({ + "to": ["first@example.test", "invalid recipient", "third@example.test"], + }) + + assert captured.value.errors()[0]["loc"] == ("to", 1) + assert "email addresses" in captured.value.errors()[0]["msg"] + + def test_report_jobs_csv_attachment_requires_recipient_export_permission() -> None: payload = ReportEmailRequest(to=["recipient@example.test"], attach_jobs_csv=True) principal = SimpleNamespace(tenant_id="tenant-1")