diff --git a/AGENTS.md b/AGENTS.md index ab10b79..50dfdf1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,11 @@ # GovOPlaN Campaign Codex Guide +## Documentation Contract + +- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior. +- Keep feature content here; `govoplan-docs` projects it without importing Campaign internals. +- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes. + ## Scope This repository owns the `campaigns` module: campaign authoring, validation, message building, attachment resolution, queue/review/send control, reports, campaign module manifest, and `@govoplan/campaign-webui`. diff --git a/README.md b/README.md index eca70a7..c93b179 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,13 @@ primitives, CSRF/API helpers, shell layout, and route rendering. Tenancy is an optional platform module for tenant administration and tenant resolver behavior. Files and mail own their respective storage and transport capabilities. +Generated EML is durable execution material, not a node-local runtime cache. +Campaign stores it through Core's shared object-storage contract under opaque +Campaign-owned keys; database job rows retain the expected size, digest, and +Message-ID. Workers resolve and verify the object before delivery. Build +failure compensates objects written before database commit, and retention keeps +the database reference when object deletion fails so cleanup can be retried. + ## Dependencies The module has one required runtime dependency: diff --git a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md index 7ad8d8a..c2c3bd5 100644 --- a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md +++ b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md @@ -134,6 +134,29 @@ production process manager. Repeat the worker-loss drill under the selected systemd, container, Kubernetes, or other production supervisor and the target Redis/SMTP infrastructure before deployment approval. +## Shared Build Artifacts + +Generated EML is stored through Core's configured object-storage backend under +opaque Campaign-owned keys. The job records expected byte size, SHA-256 digest, +and Message-ID. A worker on another node must retrieve and verify those values +before attempting delivery. + +- Do not expose object keys to ordinary campaign users or copy them into + business fields. +- A build failure deletes objects written before the database transaction can + commit. +- Retention clears metadata only after object deletion succeeds; an unavailable + backend leaves the reference in place for a later retry. +- A hard process loss between object creation and metadata commit can leave an + orphan object. Reconcile only within the Campaign build prefix and verify that + no job references the object before deleting it. +- Restore Campaign rows, object storage, and the encryption key to one + coordinated recovery point before resuming workers. + +For a scaled-runtime drill, build on one API replica, consume from another +worker, compare the stored evidence, and inject storage failures during build +and retention. + ## Reporting Checks - Partial delivery must show accepted, failed, and unknown counts separately. diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index 7da2b71..bac4181 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -28,6 +28,7 @@ from govoplan_core.core.modules import ( RoleTemplate, ) from govoplan_core.core.operations import OperationalCheckProviderRegistration +from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_core.core.postbox import ( @@ -718,6 +719,37 @@ manifest = ModuleManifest( ], }, ), + DocumentationTopic( + id="campaigns.reference.shared-build-artifacts", + title="Operate Campaign build artifacts across workers", + summary="Generated messages use shared object storage and are verified before delivery.", + body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Failed builds compensate newly written objects; retention retains metadata when deletion fails. Never copy or edit runtime object keys as business data.", + layer="evidence", + documentation_types=("admin",), + audience=("campaign_operator", "platform_operator", "release_reviewer"), + order=53, + conditions=( + DocumentationCondition( + required_modules=("campaigns",), + any_scopes=("campaigns:diagnostic:read", "system:settings:read"), + ), + ), + links=( + DocumentationLink(label="Campaign operator queue", href="/campaigns/queue", kind="runtime"), + DocumentationLink(label="Campaign delivery runbook", href="govoplan-campaign/docs/CAMPAIGN_DELIVERY_RUNBOOK.md", kind="repository"), + ), + related_modules=("files", "ops", "mail"), + metadata={ + "kind": "reference", + "route": "/campaigns/queue", + "screen": "Campaign operator queue", + "verification": "Build on one replica, deliver from another, verify object digest/size evidence, and exercise storage failure during build and retention.", + "limitations": [ + "A hard process loss between object creation and database commit can leave an orphan object until an inventory reconciler removes it.", + "Database, object storage, and encryption keys require a coordinated deployment backup and restore procedure.", + ], + }, + ), ), documentation_providers=(documentation_topics,), ownership_providers=( @@ -762,6 +794,19 @@ manifest = ModuleManifest( cache_seconds=60, ), ), + architecture=declared_module_architecture( + layer="communication_participation", + kind="domain", + maturity="vertical_slice", + documentation_ref="docs/CAMPAIGN_HANDBOOK.md", + test_ref="tests/test_execution_snapshot_integrity.py", + known_limits=("Target-environment recovery and accessibility evidence is incomplete for reference-ready maturity.",), + owned_concepts=("campaign", "campaign version", "recipient snapshot", "delivery job", "delivery report"), + non_owned_concepts=("mail transport", "postbox", "durable address directory", "file storage"), + recovery_docs=("docs/CAMPAIGN_DELIVERY_RUNBOOK.md",), + security_docs=("docs/ACCESS_EXPLANATION_COVERAGE.md",), + operations_docs=("docs/CAMPAIGN_DELIVERY_RUNBOOK.md",), + ), ) diff --git a/src/govoplan_campaign/backend/operational_checks.py b/src/govoplan_campaign/backend/operational_checks.py index 906827b..c3e5bc4 100644 --- a/src/govoplan_campaign/backend/operational_checks.py +++ b/src/govoplan_campaign/backend/operational_checks.py @@ -1,55 +1,56 @@ from __future__ import annotations -import os import secrets -from pathlib import Path +from govoplan_core.core.object_storage import ( + StorageBackendError, + configured_storage_backend, +) from govoplan_core.core.operations import OperationalCheck -from govoplan_campaign.backend.persistence.campaigns import BUILD_OUTPUT_DIR +from govoplan_core.settings import settings as core_settings +from govoplan_campaign.backend.runtime import get_settings def generated_eml_storage_check() -> OperationalCheck: - """Verify the current EML evidence path and report its node-local boundary.""" + """Verify Campaign evidence against the deployment object-store boundary.""" - root = Path(BUILD_OUTPUT_DIR) - probe = root / ".govoplan-health" / f"{secrets.token_hex(16)}.probe" + storage = configured_storage_backend(get_settings() or core_settings) + probe = f"campaign-artifacts/.health/{secrets.token_hex(16)}.probe" payload = secrets.token_bytes(64) try: - probe.parent.mkdir(parents=True, mode=0o700, exist_ok=True) - with probe.open("xb") as handle: - handle.write(payload) - handle.flush() - os.fsync(handle.fileno()) - if probe.read_bytes() != payload: + storage.put_bytes(probe, payload, content_type="application/octet-stream") + if storage.get_bytes(probe) != payload: raise OSError("generated EML persistence returned different bytes") - except OSError as exc: + except (OSError, StorageBackendError) as exc: return OperationalCheck( id="campaign.generated_eml_storage", label="Generated Campaign EML evidence", state="error", detail=( - "The generated EML evidence path failed a bounded durable-write probe " + "The generated EML object store failed a bounded write/read probe " f"({type(exc).__name__})." ), readiness_critical=True, - metrics={"backend": "node_local_filesystem"}, + metrics={"backend": storage.name}, ) finally: try: - probe.unlink(missing_ok=True) - probe.parent.rmdir() - except OSError: + storage.delete(probe) + except StorageBackendError: pass + node_local = storage.name == "local" return OperationalCheck( id="campaign.generated_eml_storage", label="Generated Campaign EML evidence", - state="warning", + state="warning" if node_local else "ok", detail=( - "Generated EML passed a local write/fsync/read probe, but remains node-local. " - "Use one shared runtime volume and include it in coordinated backups until " - "Campaign evidence is migrated to managed object storage." + "Generated EML passed the object-store write/read/delete probe. " + + ( + "The configured backend is node-local and is suitable only for a single-node profile." + if node_local + else "The configured backend is shared across application and worker nodes." + ) ), - metrics={"backend": "node_local_filesystem"}, + metrics={"backend": storage.name}, ) - diff --git a/src/govoplan_campaign/backend/persistence/campaigns.py b/src/govoplan_campaign/backend/persistence/campaigns.py index a0a9ee7..ceef032 100644 --- a/src/govoplan_campaign/backend/persistence/campaigns.py +++ b/src/govoplan_campaign/backend/persistence/campaigns.py @@ -2,18 +2,24 @@ from __future__ import annotations import copy import hashlib -import json -import os +from dataclasses import dataclass from datetime import UTC, datetime from email import policy from email.parser import BytesParser from pathlib import Path +from tempfile import TemporaryDirectory from typing import Any from uuid import uuid4 from sqlalchemy import func from sqlalchemy.orm import Session +from govoplan_core.core.object_storage import ( + StorageBackend, + StorageBackendError, + configured_storage_backend, +) +from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, @@ -52,20 +58,19 @@ from govoplan_campaign.backend.integrations import ( postbox_integration, ) from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths - -RUNTIME_DIR = Path(__file__).resolve().parents[3] / "runtime" -CAMPAIGN_SNAPSHOT_DIR = RUNTIME_DIR / "campaign_snapshots" -BUILD_OUTPUT_DIR = RUNTIME_DIR / "generated_eml" +from govoplan_campaign.backend.runtime import get_settings class CampaignPersistenceError(RuntimeError): pass -def _ensure_dirs() -> None: - for directory in (CAMPAIGN_SNAPSHOT_DIR, BUILD_OUTPUT_DIR): - directory.mkdir(parents=True, mode=0o700, exist_ok=True) - directory.chmod(0o700) +@dataclass(frozen=True, slots=True) +class _StoredEmlArtifact: + storage_key: str + size_bytes: int + sha256: str + message_id_header: str | None @@ -107,22 +112,82 @@ def load_campaign_config_from_json( return CampaignConfig.model_validate(materialized) -def _write_campaign_snapshot(version: CampaignVersion) -> Path: - _ensure_dirs() - path = CAMPAIGN_SNAPSHOT_DIR / f"{version.id}.json" - flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC - if hasattr(os, "O_NOFOLLOW"): - flags |= os.O_NOFOLLOW - descriptor = os.open(path, flags, 0o600) +def _campaign_reference_path(version: CampaignVersion) -> Path: + """Return a path anchor without persisting a second configuration copy.""" + + source_base_path = str(version.source_base_path or "").strip() + base = Path(source_base_path).expanduser().resolve() if source_base_path else Path.cwd() + return base / "campaign.json" + + +def _object_storage() -> StorageBackend: + return configured_storage_backend(get_settings() or core_settings) + + +def _persist_built_eml_artifacts( + *, + storage: StorageBackend, + tenant_id: str, + campaign_id: str, + version_id: str, + build_id: str, + built_messages: list[Any], +) -> dict[int, _StoredEmlArtifact]: + artifacts: dict[int, _StoredEmlArtifact] = {} try: - os.fchmod(descriptor, 0o600) - with os.fdopen(descriptor, "w", encoding="utf-8") as stream: - descriptor = -1 - json.dump(version.raw_json, stream, ensure_ascii=False, indent=2) - finally: - if descriptor >= 0: - os.close(descriptor) - return path + for built in built_messages: + message = built.draft + if not message.eml_path: + continue + path = Path(message.eml_path) + if not path.is_file(): + raise CampaignPersistenceError( + "Generated EML evidence is missing for entry " + f"{message.entry_index}" + ) + payload = path.read_bytes() + digest = hashlib.sha256(payload).hexdigest() + parsed = BytesParser(policy=policy.default).parsebytes(payload) + message_id = parsed.get("Message-ID") + storage_key = ( + f"campaign-artifacts/{tenant_id}/{campaign_id}/{version_id}/" + f"{build_id}/{message.entry_index:08d}-{digest}.eml" + ) + try: + storage.put_bytes( + storage_key, + payload, + content_type="message/rfc822", + ) + except StorageBackendError as exc: + raise CampaignPersistenceError( + f"Generated EML evidence could not be persisted: {exc}" + ) from exc + artifacts[message.entry_index] = _StoredEmlArtifact( + storage_key=storage_key, + size_bytes=len(payload), + sha256=digest, + message_id_header=str(message_id) if message_id else None, + ) + message.eml_path = None + message.eml_size_bytes = len(payload) + except Exception: + _delete_storage_keys( + storage, + [artifact.storage_key for artifact in artifacts.values()], + ) + raise + return artifacts + + +def _delete_storage_keys(storage: StorageBackend, keys: list[str]) -> None: + for key in keys: + try: + storage.delete(key) + except StorageBackendError: + # A committed build remains authoritative. Reconciliation can + # remove an orphaned superseded object later. + continue def _next_version_number(session: Session, campaign_id: str) -> int: @@ -143,11 +208,11 @@ def normalize_campaign_paths(raw_json: dict[str, Any], source_base_path: str | P """Resolve paths for an explicitly trusted, file-oriented import. The CLI naturally resolves relative paths against the campaign.json file. - Once the campaign is stored in the database, the JSON snapshot lives in - app/mailer/runtime/campaign_snapshots. To keep existing file-based - campaigns working, relative file paths are normalized to absolute paths at - import time when a source_base_path is known. HTTP/API callers are rejected - by ``assert_server_safe_campaign_paths`` before they can reach this helper. + Once the campaign is stored in the database, its JSON is authoritative. + To keep existing file-based campaigns working, relative file paths are + normalized to absolute paths at import time when a source_base_path is + known. HTTP/API callers are rejected by + ``assert_server_safe_campaign_paths`` before they can reach this helper. """ base = Path(source_base_path).expanduser().resolve() if source_base_path else None data = copy.deepcopy(raw_json) @@ -232,7 +297,6 @@ def create_campaign_version_from_json( campaign.current_version_id = version.id session.add(campaign) if commit: - _write_campaign_snapshot(version) session.commit() else: session.flush() @@ -299,7 +363,7 @@ def load_version_config(session: Session, version_id: str): raw_json, managed_files_available=files_integration().available, ) - path = _write_campaign_snapshot(version) + path = _campaign_reference_path(version) return version, path, load_campaign_config_from_json(session, tenant_id=campaign.tenant_id, raw_json=raw_json, campaign_id=campaign.id) @@ -391,8 +455,6 @@ def validate_campaign_version( version.workflow_state = CampaignVersionWorkflowState.APPROVED.value version.is_complete = True if lock_on_success and version.locked_at is None: - from datetime import UTC, datetime - version.locked_at = datetime.now(UTC) version.locked_by_user_id = user_id else: @@ -426,9 +488,13 @@ def _job_from_message( version_id: str, message: MessageDraft, resolved_postbox_targets: list[dict[str, Any]] | None = None, + stored_eml: _StoredEmlArtifact | None = None, ) -> CampaignJob: recipient_email = message.to[0].email if message.to else None eml_sha256, message_id_header = _eml_evidence(message.eml_path) + if stored_eml is not None: + eml_sha256 = stored_eml.sha256 + message_id_header = stored_eml.message_id_header return CampaignJob( tenant_id=tenant_id, campaign_id=campaign_id, @@ -438,8 +504,9 @@ def _job_from_message( recipient_email=recipient_email, subject=message.subject, message_id_header=message_id_header, - eml_local_path=message.eml_path, - eml_size_bytes=message.eml_size_bytes, + eml_storage_key=stored_eml.storage_key if stored_eml else None, + eml_local_path=message.eml_path if stored_eml is None else None, + eml_size_bytes=stored_eml.size_bytes if stored_eml else message.eml_size_bytes, eml_sha256=eml_sha256, build_status=message.build_status.value if hasattr(message.build_status, "value") else str(message.build_status), validation_status=_job_validation_status(message.validation_status.value), @@ -531,7 +598,18 @@ def _replace_version_jobs( version_id: str, built_messages: list[Any], postbox_targets_by_index: dict[int, list[dict[str, Any]]], -) -> list[tuple[CampaignJob, MessageDraft]]: + stored_eml_by_index: dict[int, _StoredEmlArtifact], +) -> tuple[list[tuple[CampaignJob, MessageDraft]], list[str]]: + old_storage_keys = [ + str(key) + for (key,) in session.query(CampaignJob.eml_storage_key) + .filter( + CampaignJob.campaign_version_id == version_id, + CampaignJob.eml_storage_key.is_not(None), + ) + .all() + if key + ] session.query(CampaignIssue).filter( CampaignIssue.campaign_version_id == version_id, CampaignIssue.job_id.is_not(None), @@ -547,11 +625,12 @@ def _replace_version_jobs( version_id=version_id, message=built.draft, resolved_postbox_targets=postbox_targets_by_index.get(built.draft.entry_index, []), + stored_eml=stored_eml_by_index.get(built.draft.entry_index), ) session.add(job) pairs.append((job, built.draft)) session.flush() - return pairs + return pairs, old_storage_keys def _mail_execution_profile( @@ -664,71 +743,112 @@ def build_campaign_version( raise CampaignPersistenceError("Campaign version must be successfully validated before messages are built") _ensure_version_validated_and_locked(version) - output_dir = BUILD_OUTPUT_DIR / campaign.id / version.id files = files_integration() - with files.prepared_campaign_snapshot( - session, - tenant_id=tenant_id, - campaign_id=campaign.id, - raw_json=version.raw_json if isinstance(version.raw_json, dict) else {}, - include_bytes=True, - prefix="govoplan-managed-build-", - ) as prepared: - managed_raw = load_campaign_json(prepared.path) - managed_config = load_campaign_config_from_json(session, tenant_id=tenant_id, raw_json=managed_raw, campaign_id=campaign.id) - result = build_campaign_messages(managed_config, campaign_file=prepared.path, output_dir=output_dir, write_eml=write_eml) - files.annotate_built_messages_with_managed_files(result.built_messages, prepared.managed_files_by_local_path) - entries_by_index = { - index: entry - for index, entry in enumerate( - load_campaign_entries( - managed_config, - campaign_file=prepared.path, - ), - start=1, - ) - } - resolved_postbox_targets_by_index = _resolve_built_postbox_targets( + storage = _object_storage() + build_id = uuid4().hex + with TemporaryDirectory(prefix="govoplan-campaign-build-") as output_directory: + with files.prepared_campaign_snapshot( session, tenant_id=tenant_id, - config=managed_config, + campaign_id=campaign.id, + raw_json=version.raw_json if isinstance(version.raw_json, dict) else {}, + include_bytes=True, + prefix="govoplan-managed-build-", + ) as prepared: + managed_raw = load_campaign_json(prepared.path) + managed_config = load_campaign_config_from_json( + session, + tenant_id=tenant_id, + raw_json=managed_raw, + campaign_id=campaign.id, + ) + result = build_campaign_messages( + managed_config, + campaign_file=prepared.path, + output_dir=Path(output_directory), + write_eml=write_eml, + ) + files.annotate_built_messages_with_managed_files( + result.built_messages, + prepared.managed_files_by_local_path, + ) + entries_by_index = { + index: entry + for index, entry in enumerate( + load_campaign_entries( + managed_config, + campaign_file=prepared.path, + ), + start=1, + ) + } + 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, + ) + stored_eml_by_index = _persist_built_eml_artifacts( + storage=storage, + tenant_id=tenant_id, + campaign_id=campaign.id, + version_id=version.id, + build_id=build_id, built_messages=result.built_messages, - entries_by_index=entries_by_index, ) - report_json = _campaign_build_report(result, files) - report_json["built_by_user_id"] = user_id - version.build_summary = report_json - editor_state = copy.deepcopy(version.editor_state or {}) - editor_state.pop("review_send", None) - version.editor_state = editor_state + new_storage_keys = [item.storage_key for item in stored_eml_by_index.values()] + try: + report_json = _campaign_build_report(result, files) + report_json["built_by_user_id"] = user_id + version.build_summary = report_json + editor_state = copy.deepcopy(version.editor_state or {}) + editor_state.pop("review_send", None) + version.editor_state = editor_state - job_build_pairs = _replace_version_jobs( - session, - 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, - ) - 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, - ) - _apply_campaign_build_state( - campaign, - version, - needs_review_count=result.report.needs_review_count, - blocked_count=result.report.blocked_count, - queueable_count=result.report.queueable_count, - ) + job_build_pairs, old_storage_keys = _replace_version_jobs( + session, + 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, + stored_eml_by_index=stored_eml_by_index, + ) + 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, + ) + _apply_campaign_build_state( + campaign, + version, + needs_review_count=result.report.needs_review_count, + blocked_count=result.report.blocked_count, + queueable_count=result.report.queueable_count, + ) - session.add(version) - session.add(campaign) - session.commit() + session.add(version) + session.add(campaign) + session.commit() + except Exception: + session.rollback() + _delete_storage_keys(storage, new_storage_keys) + raise + _delete_storage_keys(storage, old_storage_keys) return report_json diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index a7aed2c..b857228 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -38,7 +38,6 @@ from govoplan_campaign.backend.integrations import files_integration, mail_integ from govoplan_campaign.backend.persistence.campaigns import ( CampaignPersistenceError, _next_version_number, - _write_campaign_snapshot, normalize_campaign_paths, ) from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths @@ -250,7 +249,6 @@ def create_minimal_campaign( campaign.current_version_id = version.id session.add(campaign) if commit: - _write_campaign_snapshot(version) session.commit() else: session.flush() @@ -435,7 +433,6 @@ def _persist_forked_version( campaign.status = CampaignStatus.DRAFT.value session.add(campaign) if commit: - _write_campaign_snapshot(version) session.commit() else: session.flush() @@ -705,7 +702,6 @@ def _persist_updated_version( session.add(campaign) session.flush() if commit: - _write_campaign_snapshot(version) session.commit() else: session.flush() diff --git a/src/govoplan_campaign/backend/retention.py b/src/govoplan_campaign/backend/retention.py index d35b576..c3c13dd 100644 --- a/src/govoplan_campaign/backend/retention.py +++ b/src/govoplan_campaign/backend/retention.py @@ -9,7 +9,15 @@ from typing import Any, Callable from sqlalchemy.orm import Session +from govoplan_core.core.object_storage import ( + StorageBackend, + StorageBackendError, + StorageObjectMissing, + configured_storage_backend, +) +from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import CampaignJob, CampaignVersion, JobImapStatus, JobQueueStatus +from govoplan_campaign.backend.runtime import get_settings FINAL_VERSION_STATES = { "completed", @@ -111,8 +119,16 @@ def _apply_eml_retention( dry_run: bool, now: datetime, policy_for_campaign_id: Callable[[str | None], object], + storage: StorageBackend | None = None, ) -> dict[str, int]: - result = {"eligible": 0, "metadata_cleared": 0, "files_deleted": 0, "files_missing": 0, "skipped_not_final": 0} + result = { + "eligible": 0, + "metadata_cleared": 0, + "files_deleted": 0, + "files_missing": 0, + "delete_failed": 0, + "skipped_not_final": 0, + } jobs = ( session.query(CampaignJob) .filter((CampaignJob.eml_local_path.is_not(None)) | (CampaignJob.eml_storage_key.is_not(None))) @@ -137,6 +153,21 @@ def _apply_eml_retention( result["eligible"] += 1 if dry_run: continue + if job.eml_storage_key: + active_storage = storage or configured_storage_backend( + get_settings() or core_settings + ) + try: + if active_storage.exists(job.eml_storage_key): + active_storage.delete(job.eml_storage_key) + result["files_deleted"] += 1 + else: + result["files_missing"] += 1 + except StorageObjectMissing: + result["files_missing"] += 1 + except StorageBackendError: + result["delete_failed"] += 1 + continue if job.eml_local_path: path = Path(job.eml_local_path) if path.exists(): diff --git a/src/govoplan_campaign/backend/route_support.py b/src/govoplan_campaign/backend/route_support.py index cd633ec..1c45bcd 100644 --- a/src/govoplan_campaign/backend/route_support.py +++ b/src/govoplan_campaign/backend/route_support.py @@ -645,17 +645,9 @@ def _clear_current_version_mail_profile_for_owner_transfer( def _write_current_version_snapshot_if_available(version: CampaignVersion) -> None: - try: - from govoplan_campaign.backend.persistence.campaigns import ( - _write_campaign_snapshot, - ) - - _write_campaign_snapshot(version) - except Exception: - # The database state is authoritative for the WebUI. Snapshot writing is - # best-effort here because ownership changes should not fail due to an - # unavailable local runtime directory. - return + # Kept as a compatibility no-op for callers outside this package. Campaign + # JSON is database-authoritative and no longer mirrored onto an API node. + del version def bounded_query_rows(query, *, limit: int, label: str): diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index ffc68c6..f53531f 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -16,6 +16,10 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider +from govoplan_core.core.object_storage import ( + StorageBackendError, + configured_storage_backend, +) from govoplan_core.audit.logging import audit_event from govoplan_core.security.redaction import redact_secret_values from govoplan_core.settings import settings as core_settings @@ -53,7 +57,7 @@ from govoplan_campaign.backend.sending.postbox_delivery import ( PostboxChannelOutcome, deliver_campaign_job_to_postboxes, ) -from govoplan_campaign.backend.runtime import get_registry +from govoplan_campaign.backend.runtime import get_registry, get_settings from govoplan_campaign.backend.integrations import ( ImapAppendError, ImapConfigurationError, @@ -2389,6 +2393,17 @@ def _verify_eml_evidence(job: CampaignJob, payload: bytes) -> None: def _load_eml_bytes_for_job(job: CampaignJob) -> bytes: + if job.eml_storage_key: + try: + payload = configured_storage_backend( + get_settings() or core_settings + ).get_bytes(job.eml_storage_key) + except StorageBackendError as exc: + raise SendJobError( + f"Generated EML object could not be read for job {job.id}: {exc}" + ) from exc + _verify_eml_evidence(job, payload) + return payload if job.eml_local_path: path = Path(job.eml_local_path) if not path.exists(): @@ -2396,7 +2411,7 @@ def _load_eml_bytes_for_job(job: CampaignJob) -> bytes: payload = path.read_bytes() _verify_eml_evidence(job, payload) return payload - raise SendJobError("Only local EML paths are supported for sending in this implementation step") + raise SendJobError("Generated EML evidence is not available for this job") def _load_eml_for_job(job: CampaignJob): diff --git a/tests/test_campaign_optimistic_concurrency.py b/tests/test_campaign_optimistic_concurrency.py index 7353ec1..e69d73d 100644 --- a/tests/test_campaign_optimistic_concurrency.py +++ b/tests/test_campaign_optimistic_concurrency.py @@ -92,9 +92,6 @@ class CampaignOptimisticConcurrencyTests(unittest.TestCase): "govoplan_campaign.backend.persistence.versions._updated_runtime_json", side_effect=lambda _session, **kwargs: kwargs["raw_json"], ), - patch( - "govoplan_campaign.backend.persistence.versions._write_campaign_snapshot" - ), ): return update_campaign_version( session, diff --git a/tests/test_operational_checks.py b/tests/test_operational_checks.py index cfe4041..482840d 100644 --- a/tests/test_operational_checks.py +++ b/tests/test_operational_checks.py @@ -2,17 +2,17 @@ from __future__ import annotations from unittest.mock import patch +from govoplan_core.core.object_storage import LocalFilesystemStorageBackend from govoplan_campaign.backend.operational_checks import generated_eml_storage_check def test_generated_eml_probe_reports_node_local_boundary(tmp_path) -> None: with patch( - "govoplan_campaign.backend.operational_checks.BUILD_OUTPUT_DIR", - tmp_path, + "govoplan_campaign.backend.operational_checks.configured_storage_backend", + return_value=LocalFilesystemStorageBackend(tmp_path), ): result = generated_eml_storage_check() assert result.state == "warning" assert "node-local" in result.detail assert list(tmp_path.rglob("*.probe")) == [] - diff --git a/tests/test_response_security.py b/tests/test_response_security.py index 564b655..929c54f 100644 --- a/tests/test_response_security.py +++ b/tests/test_response_security.py @@ -1,6 +1,5 @@ from __future__ import annotations -import stat import unittest from datetime import UTC, datetime from types import SimpleNamespace @@ -307,20 +306,18 @@ def test_diagnostics_permission_is_operator_only_by_default() -> None: assert "campaigns:diagnostic:read" not in role_permissions["campaign_reviewer"] -def test_campaign_snapshot_with_inline_credentials_is_owner_only(tmp_path, monkeypatch) -> None: +def test_database_campaign_config_is_not_mirrored_to_node_local_disk(tmp_path) -> None: from govoplan_campaign.backend.persistence import campaigns as persistence - snapshots = tmp_path / "snapshots" - output = tmp_path / "generated" - monkeypatch.setattr(persistence, "CAMPAIGN_SNAPSHOT_DIR", snapshots) - monkeypatch.setattr(persistence, "BUILD_OUTPUT_DIR", output) - path = persistence._write_campaign_snapshot( # type: ignore[arg-type] - SimpleNamespace(id="version-secret", raw_json={"server": {"smtp": {"password": "secret"}}}) + path = persistence._campaign_reference_path( # type: ignore[arg-type] + SimpleNamespace( + source_base_path=str(tmp_path), + raw_json={"server": {"smtp": {"password": "secret"}}}, + ) ) - assert stat.S_IMODE(path.stat().st_mode) == 0o600 - assert stat.S_IMODE(snapshots.stat().st_mode) == 0o700 - assert stat.S_IMODE(output.stat().st_mode) == 0o700 + assert path == tmp_path / "campaign.json" + assert list(tmp_path.iterdir()) == [] class ResponseSecurityTests(unittest.TestCase): diff --git a/tests/test_server_path_security.py b/tests/test_server_path_security.py index 135dc9f..8ebdd65 100644 --- a/tests/test_server_path_security.py +++ b/tests/test_server_path_security.py @@ -79,13 +79,13 @@ class ServerCampaignPathSecurityTests(unittest.TestCase): "govoplan_campaign.backend.persistence.campaigns.files_integration", return_value=SimpleNamespace(available=True), ), - patch("govoplan_campaign.backend.persistence.campaigns._write_campaign_snapshot") as write_snapshot, + patch("govoplan_campaign.backend.persistence.campaigns._campaign_reference_path") as reference_path, patch("govoplan_campaign.backend.persistence.campaigns.load_campaign_config_from_json") as load_config, ): with self.assertRaisesRegex(CampaignPathSecurityError, "template source paths"): load_version_config(session, "version-1") # type: ignore[arg-type] - write_snapshot.assert_not_called() + reference_path.assert_not_called() load_config.assert_not_called() def test_source_base_symlink_escape_is_rejected(self) -> None: diff --git a/tests/test_shared_artifact_storage.py b/tests/test_shared_artifact_storage.py new file mode 100644 index 0000000..3c8b935 --- /dev/null +++ b/tests/test_shared_artifact_storage.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import hashlib +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from govoplan_core.core.object_storage import ( + LocalFilesystemStorageBackend, + StorageBackendError, +) +from govoplan_campaign.backend.persistence.campaigns import ( + CampaignPersistenceError, + _persist_built_eml_artifacts, +) +from govoplan_campaign.backend.retention import _apply_eml_retention +from govoplan_campaign.backend.sending.jobs import _load_eml_bytes_for_job + + +class _FailingStorage: + name = "test" + + def __init__(self) -> None: + self.objects: dict[str, bytes] = {} + self.deleted: list[str] = [] + + def put_bytes(self, key: str, data: bytes, **_kwargs) -> None: + if self.objects: + raise StorageBackendError("object service unavailable") + self.objects[key] = data + + def delete(self, key: str) -> None: + self.deleted.append(key) + self.objects.pop(key, None) + + +def test_partial_campaign_artifact_write_is_compensated(tmp_path: Path) -> None: + first = tmp_path / "first.eml" + second = tmp_path / "second.eml" + first.write_bytes(b"Message-ID: \r\n\r\nfirst") + second.write_bytes(b"Message-ID: \r\n\r\nsecond") + messages = [ + SimpleNamespace( + draft=SimpleNamespace( + entry_index=1, + eml_path=str(first), + eml_size_bytes=None, + ) + ), + SimpleNamespace( + draft=SimpleNamespace( + entry_index=2, + eml_path=str(second), + eml_size_bytes=None, + ) + ), + ] + storage = _FailingStorage() + + with pytest.raises(CampaignPersistenceError, match="could not be persisted"): + _persist_built_eml_artifacts( + storage=storage, # type: ignore[arg-type] + tenant_id="tenant-1", + campaign_id="campaign-1", + version_id="version-1", + build_id="build-1", + built_messages=messages, + ) + + assert storage.objects == {} + assert len(storage.deleted) == 1 + + +def test_worker_loads_and_verifies_campaign_eml_from_object_storage( + tmp_path: Path, +) -> None: + payload = b"Message-ID: \r\n\r\nbody" + storage = LocalFilesystemStorageBackend(tmp_path) + storage.put_bytes("campaign-artifacts/message.eml", payload) + job = SimpleNamespace( + id="job-1", + eml_storage_key="campaign-artifacts/message.eml", + eml_local_path=None, + eml_size_bytes=len(payload), + eml_sha256=hashlib.sha256(payload).hexdigest(), + message_id_header="", + ) + + with patch( + "govoplan_campaign.backend.sending.jobs.configured_storage_backend", + return_value=storage, + ): + assert _load_eml_bytes_for_job(job) == payload + + +def test_retention_preserves_reference_when_object_store_is_unavailable() -> None: + job = SimpleNamespace( + campaign_id="campaign-1", + updated_at=datetime.now(timezone.utc) - timedelta(days=10), + queue_status="draft", + send_status="smtp_accepted", + imap_status="appended", + eml_local_path=None, + eml_storage_key="campaign-artifacts/message.eml", + ) + session = MagicMock() + session.query.return_value.filter.return_value.order_by.return_value.all.return_value = [ + job + ] + storage = MagicMock() + storage.exists.side_effect = StorageBackendError("object service unavailable") + + result = _apply_eml_retention( + session, + dry_run=False, + now=datetime.now(timezone.utc), + policy_for_campaign_id=lambda _campaign_id: SimpleNamespace( + generated_eml_retention_days=1 + ), + storage=storage, + ) + + assert result["delete_failed"] == 1 + assert result["metadata_cleared"] == 0 + assert job.eml_storage_key == "campaign-artifacts/message.eml" + session.add.assert_not_called()