feat(campaign): expose attachment linking and single send

src/govoplan_campaign/backend/integrations.py

@@ -50,6 +50,7 @@ class _PreparedCampaignSnapshot:
         self.raw_json = raw_json
         self.managed_files_by_local_path: dict[str, Any] = {}
         self.shared_assets: list[Any] = []
+        self.candidate_assets: list[Any] = []
 
     def cleanup(self) -> None:
         shutil.rmtree(self._directory, ignore_errors=True)
@@ -107,6 +108,11 @@ class FilesCampaignIntegration:
             raise OptionalModuleUnavailable("Files module is not available")
         return self._delegate.current_version_and_blob(session, asset)
 
+    def share_assets_with_campaign(self, session: Any, **kwargs: Any) -> list[dict[str, Any]]:
+        if self._delegate is None:
+            raise OptionalModuleUnavailable("Files module is not available")
+        return self._delegate.share_assets_with_campaign(session, **kwargs)
+
     def mark_job_attachment_uses_sent(self, session: Any, job: Any) -> None:
         if self._delegate is not None:
             self._delegate.mark_job_attachment_uses_sent(session, job)

src/govoplan_campaign/backend/router.py

@@ -11,7 +11,9 @@ from sqlalchemy.orm import Session
 from pydantic import BaseModel, Field
 
 from govoplan_campaign.backend.schemas import (
+    AppendSentRequest,
     BuildCampaignRequest,
+    CampaignActionResponse,
     CampaignDeltaResponse,
     CampaignCreateRequest,
     CampaignUpdateRequest,
@@ -37,6 +39,7 @@ from govoplan_campaign.backend.schemas import (
     CampaignJobsDeltaResponse,
     CampaignJobDetailResponse,
     CampaignRetryJobsRequest,
+    CampaignSendJobRequest,
     CampaignSendUnattemptedRequest,
     CampaignResolveOutcomeRequest,
     CampaignListResponse,
@@ -53,6 +56,12 @@ from govoplan_campaign.backend.schemas import (
     ValidateCampaignRequest,
     ReportEmailRequest,
     ReportEmailResponse,
+    MockCampaignSendRequest,
+    MockCampaignSendResponse,
+    QueueCampaignRequest,
+    QueueCampaignResponse,
+    SendCampaignNowRequest,
+    SendCampaignNowResponse,
 )
 from govoplan_core.auth import ApiPrincipal, has_scope, require_scope
 from govoplan_core.audit.logging import audit_from_principal
@@ -118,7 +127,21 @@ from govoplan_campaign.backend.persistence.versions import (
     update_campaign_review_state,
     validate_campaign_partial,
 )
-from govoplan_campaign.backend.sending.execution import clear_execution_snapshot
+from govoplan_campaign.backend.dev.mock_campaign import MockCampaignSendError, run_mock_campaign_send
+from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, clear_execution_snapshot
+from govoplan_campaign.backend.sending.jobs import (
+    QueueingError,
+    cancel_campaign_jobs,
+    enqueue_pending_imap_appends,
+    pause_campaign_jobs,
+    queue_campaign_jobs,
+    queue_failed_jobs_for_retry,
+    queue_unattempted_jobs,
+    reconcile_job_outcome,
+    resume_campaign_jobs,
+    send_campaign_now,
+    send_single_campaign_job,
+)
 
 router = APIRouter(prefix="/campaigns", tags=["campaigns"])
 
@@ -1728,6 +1751,7 @@ def validate_version(
 ):
     _get_version_for_principal(session, version_id, principal, write=True)
     _require_permission(principal, "campaigns:recipient:read")
+    payload = payload or ValidateCampaignRequest()
     try:
         version = _get_version_for_tenant(session, version_id, principal.tenant_id)
         _require_mail_profile_use_if_needed(principal, version.raw_json if isinstance(version.raw_json, dict) else {})
@@ -1736,11 +1760,37 @@ def validate_version(
                 status_code=status.HTTP_409_CONFLICT,
                 detail="This version has a user lock or final delivery lock and cannot be validated. Remove a temporary lock or create an editable copy.",
             )
+        link_result: CampaignAttachmentLinkMatchesResponse | None = None
+        if payload.check_files and payload.link_unshared_matches:
+            _require_permission(principal, "files:file:share")
+            campaign = _get_campaign_for_tenant(session, version.campaign_id, principal.tenant_id)
+            link_result = _link_campaign_attachment_matches(
+                session,
+                principal,
+                campaign=campaign,
+                version=version,
+                raw=version.raw_json if isinstance(version.raw_json, dict) else {},
+                dry_run=False,
+            )
+            audit_from_principal(
+                session,
+                principal,
+                action="campaign.attachment_matches_linked",
+                object_type="campaign_version",
+                object_id=version_id,
+                details={
+                    "matched_file_count": link_result.matched_file_count,
+                    "already_linked_file_count": link_result.already_linked_file_count,
+                    "linked_file_count": link_result.linked_file_count,
+                    "during_validation": True,
+                },
+                commit=True,
+            )
         result = validate_campaign_version(
             session,
             tenant_id=principal.tenant_id,
             version_id=version_id,
-            check_files=payload.check_files if payload else False,
+            check_files=payload.check_files,
             user_id=principal.user.id,
         )
         audit_from_principal(
@@ -1749,7 +1799,12 @@ def validate_version(
             action="campaign.validated",
             object_type="campaign_version",
             object_id=version_id,
-            details={"check_files": payload.check_files if payload else False, "ok": result.get("ok")},
+            details={
+                "check_files": payload.check_files,
+                "link_unshared_matches": payload.link_unshared_matches,
+                "linked_file_count": link_result.linked_file_count if link_result else 0,
+                "ok": result.get("ok"),
+            },
             commit=True,
         )
         return result
@@ -2685,30 +2740,6 @@ def revoke_campaign_share(
     return None
 
 # Queue / delivery control -------------------------------------------------
-from govoplan_campaign.backend.schemas import (
-    AppendSentRequest,
-    CampaignActionResponse,
-    QueueCampaignRequest,
-    QueueCampaignResponse,
-    SendCampaignNowRequest,
-    SendCampaignNowResponse,
-    MockCampaignSendRequest,
-    MockCampaignSendResponse,
-)
-from govoplan_campaign.backend.dev.mock_campaign import MockCampaignSendError, run_mock_campaign_send
-from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError
-from govoplan_campaign.backend.sending.jobs import (
-    QueueingError,
-    cancel_campaign_jobs,
-    enqueue_pending_imap_appends,
-    pause_campaign_jobs,
-    queue_campaign_jobs,
-    queue_failed_jobs_for_retry,
-    queue_unattempted_jobs,
-    reconcile_job_outcome,
-    resume_campaign_jobs,
-    send_campaign_now,
-)
 
 
 @router.post("/{campaign_id}/queue", response_model=QueueCampaignResponse)
@@ -2818,6 +2849,45 @@ def send_unattempted_campaign_jobs(
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
 
 
+@router.post("/{campaign_id}/jobs/{job_id}/send", response_model=CampaignActionResponse)
+def send_single_campaign_job_endpoint(
+    campaign_id: str,
+    job_id: str,
+    payload: CampaignSendJobRequest | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:send")),
+):
+    _get_campaign_for_principal(session, campaign_id, principal, write=True)
+    _require_permission(principal, "campaigns:recipient:read")
+    payload = payload or CampaignSendJobRequest()
+    _require_campaign_profile_use_if_needed(session, principal, campaign_id, None)
+    try:
+        result = send_single_campaign_job(
+            session,
+            tenant_id=principal.tenant_id,
+            campaign_id=campaign_id,
+            job_id=job_id,
+            include_warnings=payload.include_warnings,
+            dry_run=payload.dry_run,
+            use_rate_limit=payload.use_rate_limit,
+            enqueue_imap_task=payload.enqueue_imap_task,
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="campaign.single_message_sent" if not payload.dry_run else "campaign.single_message_send_dry_run",
+            object_type="campaign_job",
+            object_id=job_id,
+            details=result,
+            commit=True,
+        )
+        return CampaignActionResponse(result=result)
+    except (QueueingError, ExecutionSnapshotError) as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+
+
 @router.post("/{campaign_id}/jobs/{job_id}/resolve-outcome", response_model=CampaignActionResponse)
 def resolve_campaign_job_outcome(
     campaign_id: str,
@@ -3056,6 +3126,7 @@ def append_sent(
 
 class CampaignAttachmentPreviewRequest(BaseModel):
     include_unmatched: bool = True
+    include_unlinked_candidates: bool = False
     campaign_json: dict[str, object] | None = None
 
 
@@ -3063,10 +3134,31 @@ class CampaignAttachmentPreviewResponse(BaseModel):
     campaign_id: str
     version_id: str
     shared_file_count: int
+    candidate_file_count: int = 0
+    matched_file_count: int = 0
+    linked_file_count: int = 0
+    unlinked_file_count: int = 0
     rules: list[dict[str, object]] = Field(default_factory=list)
+    linkable_files: list[dict[str, object]] = Field(default_factory=list)
     unused_shared_files: list[dict[str, object]] = Field(default_factory=list)
 
 
+class CampaignAttachmentLinkMatchesRequest(BaseModel):
+    campaign_json: dict[str, object] | None = None
+    dry_run: bool = False
+
+
+class CampaignAttachmentLinkMatchesResponse(BaseModel):
+    campaign_id: str
+    version_id: str
+    matched_file_count: int
+    already_linked_file_count: int
+    linked_file_count: int
+    dry_run: bool = False
+    linked_files: list[dict[str, object]] = Field(default_factory=list)
+    linkable_files: list[dict[str, object]] = Field(default_factory=list)
+
+
 def _file_preview(session: Session, asset) -> dict[str, object]:
     version, blob = files_integration().current_version_and_blob(session, asset)
     return {
@@ -3080,50 +3172,12 @@ def _file_preview(session: Session, asset) -> dict[str, object]:
         "checksum_sha256": blob.checksum_sha256,
         "size_bytes": blob.size_bytes,
         "content_type": blob.content_type,
+        "linked_to_campaign": True,
     }
 
 
-@router.post("/{campaign_id}/versions/{version_id}/attachments/preview", response_model=CampaignAttachmentPreviewResponse)
+def _managed_preview_file(item: dict[str, object]) -> dict[str, object]:
-def preview_campaign_attachments(
+    return {
-    campaign_id: str,
-    version_id: str,
-    payload: CampaignAttachmentPreviewRequest | None = None,
-    session: Session = Depends(get_session),
-    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
-):
-    _get_campaign_for_principal(session, campaign_id, principal)
-    _require_permission(principal, "campaigns:recipient:read")
-    campaign = _get_campaign_for_tenant(session, campaign_id, principal.tenant_id)
-    version = _get_version_for_tenant(session, version_id, principal.tenant_id)
-    if version.campaign_id != campaign.id:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign version not found")
-
-    payload = payload or CampaignAttachmentPreviewRequest()
-    raw = payload.campaign_json if isinstance(payload.campaign_json, dict) else version.raw_json
-    raw = raw if isinstance(raw, dict) else {}
-    _require_mail_profile_use_if_needed(principal, raw)
-
-    files = files_integration()
-    with files.prepared_campaign_snapshot(
-        session,
-        tenant_id=principal.tenant_id,
-        campaign_id=campaign.id,
-        raw_json=raw,
-        include_bytes=False,
-        prefix="govoplan-managed-preview-",
-    ) as prepared:
-        prepared_raw = load_campaign_json(prepared.path)
-        config = load_campaign_config_from_json(session, tenant_id=principal.tenant_id, raw_json=prepared_raw, campaign_id=campaign.id)
-        report = resolve_campaign_attachments(config, campaign_file=prepared.path)
-        rules: list[dict[str, object]] = []
-        matched_asset_ids: set[str] = set()
-
-        for entry in report.entries:
-            for attachment in entry.attachments:
-                managed_matches = files.managed_match_payloads(attachment.matches, prepared.managed_files_by_local_path)
-                matched_asset_ids.update(str(item["asset_id"]) for item in managed_matches)
-                matches: list[dict[str, object]] = [
-                    {
         "id": item["asset_id"],
         "version_id": item["version_id"],
         "blob_id": item["blob_id"],
@@ -3134,9 +3188,53 @@ def preview_campaign_attachments(
         "checksum_sha256": item["checksum_sha256"],
         "size_bytes": item["size_bytes"],
         "content_type": item["content_type"],
+        "linked_to_campaign": bool(item.get("linked_to_campaign", True)),
     }
-                    for item in managed_matches
+
-                ]
+
+def _attachment_preview_for_version(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    campaign: Campaign,
+    version: CampaignVersion,
+    raw: dict[str, object],
+    include_unmatched: bool,
+    include_unlinked_candidates: bool,
+) -> CampaignAttachmentPreviewResponse:
+    files = files_integration()
+    with files.prepared_campaign_snapshot(
+        session,
+        tenant_id=principal.tenant_id,
+        campaign_id=campaign.id,
+        raw_json=raw,
+        include_bytes=False,
+        prefix="govoplan-managed-preview-",
+        include_unlinked_candidates=include_unlinked_candidates,
+        user_id=principal.user.id,
+        is_admin=has_scope(principal, "files:file:admin"),
+    ) as prepared:
+        prepared_raw = load_campaign_json(prepared.path)
+        config = load_campaign_config_from_json(session, tenant_id=principal.tenant_id, raw_json=prepared_raw, campaign_id=campaign.id)
+        report = resolve_campaign_attachments(config, campaign_file=prepared.path)
+        rules: list[dict[str, object]] = []
+        matched_asset_ids: set[str] = set()
+        linked_asset_ids: set[str] = set()
+        linkable_by_id: dict[str, dict[str, object]] = {}
+
+        for entry in report.entries:
+            for attachment in entry.attachments:
+                managed_matches = files.managed_match_payloads(attachment.matches, prepared.managed_files_by_local_path)
+                matches: list[dict[str, object]] = []
+                for item in managed_matches:
+                    asset_id = str(item["asset_id"])
+                    matched_asset_ids.add(asset_id)
+                    if bool(item.get("linked_to_campaign", True)):
+                        linked_asset_ids.add(asset_id)
+                    preview = _managed_preview_file(item)
+                    matches.append(preview)
+                    if not preview["linked_to_campaign"]:
+                        linkable_by_id.setdefault(asset_id, preview)
                 if not matches:
                     matches = [
                         {
@@ -3145,6 +3243,7 @@ def preview_campaign_attachments(
                             "filename": match.rsplit("/", 1)[-1].rsplit("\\", 1)[-1],
                             "owner_type": "legacy",
                             "owner_id": "",
+                            "linked_to_campaign": True,
                         }
                         for match in attachment.matches
                     ]
@@ -3167,6 +3266,8 @@ def preview_campaign_attachments(
                     "zip_filename": attachment.zip_filename,
                     "matches": matches,
                     "match_count": len(matches),
+                    "linked_match_count": sum(1 for match in matches if bool(match.get("linked_to_campaign", True))),
+                    "unlinked_match_count": sum(1 for match in matches if not bool(match.get("linked_to_campaign", True))),
                     "issues": [issue.model_dump(mode="json") for issue in attachment.issues],
                 })
 
@@ -3175,6 +3276,140 @@ def preview_campaign_attachments(
             campaign_id=campaign.id,
             version_id=version.id,
             shared_file_count=len(prepared.shared_assets),
+            candidate_file_count=len(getattr(prepared, "candidate_assets", prepared.shared_assets)),
+            matched_file_count=len(matched_asset_ids),
+            linked_file_count=len(linked_asset_ids),
+            unlinked_file_count=len(linkable_by_id),
             rules=rules,
-            unused_shared_files=[_file_preview(session, asset) for asset in unused] if payload.include_unmatched else [],
+            linkable_files=list(linkable_by_id.values()),
+            unused_shared_files=[_file_preview(session, asset) for asset in unused] if include_unmatched else [],
         )
+
+
+def _link_campaign_attachment_matches(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    campaign: Campaign,
+    version: CampaignVersion,
+    raw: dict[str, object],
+    dry_run: bool = False,
+) -> CampaignAttachmentLinkMatchesResponse:
+    preview = _attachment_preview_for_version(
+        session,
+        principal,
+        campaign=campaign,
+        version=version,
+        raw=raw,
+        include_unmatched=False,
+        include_unlinked_candidates=True,
+    )
+    file_ids = [str(item.get("id") or "") for item in preview.linkable_files if item.get("id")]
+    linked_files: list[dict[str, object]] = []
+    if file_ids and not dry_run:
+        files = files_integration()
+        shares = files.share_assets_with_campaign(
+            session,
+            tenant_id=principal.tenant_id,
+            campaign_id=campaign.id,
+            file_ids=file_ids,
+            user_id=principal.user.id,
+            is_admin=has_scope(principal, "files:file:admin"),
+        )
+        share_by_asset_id = {str(item.get("file_asset_id") or ""): item for item in shares}
+        linked_files = [
+            {**item, "share": share_by_asset_id.get(str(item.get("id") or ""))}
+            for item in preview.linkable_files
+        ]
+    return CampaignAttachmentLinkMatchesResponse(
+        campaign_id=campaign.id,
+        version_id=version.id,
+        matched_file_count=preview.matched_file_count,
+        already_linked_file_count=preview.linked_file_count,
+        linked_file_count=0 if dry_run else len(file_ids),
+        dry_run=dry_run,
+        linked_files=linked_files,
+        linkable_files=preview.linkable_files,
+    )
+
+
+@router.post("/{campaign_id}/versions/{version_id}/attachments/preview", response_model=CampaignAttachmentPreviewResponse)
+def preview_campaign_attachments(
+    campaign_id: str,
+    version_id: str,
+    payload: CampaignAttachmentPreviewRequest | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:recipient:read")
+    campaign = _get_campaign_for_tenant(session, campaign_id, principal.tenant_id)
+    version = _get_version_for_tenant(session, version_id, principal.tenant_id)
+    if version.campaign_id != campaign.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign version not found")
+
+    payload = payload or CampaignAttachmentPreviewRequest()
+    raw = payload.campaign_json if isinstance(payload.campaign_json, dict) else version.raw_json
+    raw = raw if isinstance(raw, dict) else {}
+    _require_mail_profile_use_if_needed(principal, raw)
+    return _attachment_preview_for_version(
+        session,
+        principal,
+        campaign=campaign,
+        version=version,
+        raw=raw,
+        include_unmatched=payload.include_unmatched,
+        include_unlinked_candidates=payload.include_unlinked_candidates,
+    )
+
+
+@router.post("/{campaign_id}/versions/{version_id}/attachments/link-matches", response_model=CampaignAttachmentLinkMatchesResponse)
+def link_campaign_attachment_matches(
+    campaign_id: str,
+    version_id: str,
+    payload: CampaignAttachmentLinkMatchesRequest | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:validate")),
+):
+    _require_permission(principal, "files:file:share")
+    _get_campaign_for_principal(session, campaign_id, principal, write=True)
+    _require_permission(principal, "campaigns:recipient:read")
+    campaign = _get_campaign_for_tenant(session, campaign_id, principal.tenant_id)
+    version = _get_version_for_tenant(session, version_id, principal.tenant_id)
+    if version.campaign_id != campaign.id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign version not found")
+    if is_user_locked_version(version) or is_version_final_locked(version):
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Locked campaign versions cannot link new attachment files")
+    payload = payload or CampaignAttachmentLinkMatchesRequest()
+    raw = payload.campaign_json if isinstance(payload.campaign_json, dict) else version.raw_json
+    raw = raw if isinstance(raw, dict) else {}
+    _require_mail_profile_use_if_needed(principal, raw)
+    try:
+        result = _link_campaign_attachment_matches(
+            session,
+            principal,
+            campaign=campaign,
+            version=version,
+            raw=raw,
+            dry_run=payload.dry_run,
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="campaign.attachment_matches_linked" if not payload.dry_run else "campaign.attachment_matches_link_previewed",
+            object_type="campaign_version",
+            object_id=version_id,
+            details={
+                "matched_file_count": result.matched_file_count,
+                "already_linked_file_count": result.already_linked_file_count,
+                "linked_file_count": result.linked_file_count,
+                "dry_run": result.dry_run,
+            },
+            commit=True,
+        )
+        return result
+    except HTTPException:
+        raise
+    except Exception as exc:
+        session.rollback()
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc

src/govoplan_campaign/backend/schemas.py

@@ -371,6 +371,15 @@ class CampaignSendUnattemptedRequest(BaseModel):
     dry_run: bool = False
 
 
+class CampaignSendJobRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    include_warnings: bool = True
+    dry_run: bool = False
+    use_rate_limit: bool = True
+    enqueue_imap_task: bool = False
+
+
 class CampaignResolveOutcomeRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -382,6 +391,7 @@ class ValidateCampaignRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     check_files: bool = False
+    link_unshared_matches: bool = False
 
 
 class BuildCampaignRequest(BaseModel):