Files
govoplan-campaign/src/govoplan_campaign/backend/persistence/versions.py
T

1563 lines
51 KiB
Python

from __future__ import annotations
import copy
import hashlib
import json
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4
from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import StaleDataError
from govoplan_core.core.concurrency import (
RevisionConflictError,
strong_resource_etag,
)
from govoplan_campaign.backend.db.models import (
Campaign,
CampaignIssue,
CampaignStatus,
CampaignVersion,
CampaignVersionFlow,
CampaignVersionWorkflowState,
CampaignJob,
JobSendStatus,
)
from govoplan_campaign.backend.sending.execution import clear_execution_snapshot
from govoplan_campaign.backend.campaign.mail_profile_boundary import (
campaign_editor_state_for_edit,
campaign_mail_profile_boundary_violations,
campaign_mail_profile_id,
assert_campaign_uses_mail_profile_reference,
public_campaign_mail_server,
validate_campaign_editor_state,
)
from govoplan_campaign.backend.integrations import files_integration, mail_integration
from govoplan_campaign.backend.persistence.campaigns import (
CampaignPersistenceError,
_next_version_number,
normalize_campaign_paths,
)
from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths
class LockedCampaignVersionError(CampaignPersistenceError):
"""Raised when a caller tries to edit an immutable campaign version."""
@dataclass(slots=True)
class _PartialValidationCollector:
section: str | None
issues: list[dict[str, Any]] = field(default_factory=list)
def issue(
self, severity: str, section: str, field: str, code: str, message: str
) -> None:
if self.section is None or self.section == section:
self.issues.append(
{
"severity": severity,
"section": section,
"field": field,
"code": code,
"message": message,
}
)
def result(self) -> dict[str, Any]:
return {
"ok": not any(item["severity"] == "error" for item in self.issues),
"section": self.section,
"error_count": sum(
1 for item in self.issues if item["severity"] == "error"
),
"warning_count": sum(
1 for item in self.issues if item["severity"] == "warning"
),
"info_count": sum(1 for item in self.issues if item["severity"] == "info"),
"issues": self.issues,
}
def _require_campaign(session: Session, campaign_id: str) -> Campaign:
campaign = session.get(Campaign, campaign_id)
if campaign is None:
raise CampaignPersistenceError(f"Campaign not found: {campaign_id}")
return campaign
USER_LOCK_TEMPORARY = "temporary"
USER_LOCK_PERMANENT = "permanent"
USER_LOCK_STATES = {USER_LOCK_TEMPORARY, USER_LOCK_PERMANENT}
def campaign_version_user_lock_state(version: CampaignVersion) -> str | None:
"""Return the explicit user-lock state with backwards compatibility.
Older databases represented a permanent user lock only through
published_at. Treat those rows as permanent until the migration has
backfilled the explicit state.
"""
state = getattr(version, "user_lock_state", None)
if state in USER_LOCK_STATES:
return state
if version.published_at:
return USER_LOCK_PERMANENT
return None
def minimal_campaign_json(
*, external_id: str, name: str, description: str | None = None
) -> dict[str, Any]:
"""Return a WebUI-friendly starter campaign JSON.
It is intentionally usable as an editable working copy. It contains the
main sections the UI expects, but it may still be incomplete from the
strict send/build perspective until the user configures recipients,
template and sender details.
"""
return {
"version": "1.0",
"campaign": {
"id": external_id,
"name": name,
"description": description or "",
"mode": "draft",
},
"fields": [],
"global_values": {},
"server": {},
"recipients": {
"from": [],
"allow_individual_from": False,
"to": [],
"allow_individual_to": True,
"cc": [],
"allow_individual_cc": False,
"bcc": [],
"allow_individual_bcc": False,
"reply_to": [],
"allow_individual_reply_to": False,
"bounce_to": [],
"allow_individual_bounce_to": False,
"disposition_notification_to": [],
"allow_individual_disposition_notification_to": False,
},
"template": {
"subject": "",
"text": "",
"html": None,
},
"attachments": {
"base_path": ".",
"base_paths": [
{
"id": "default",
"name": "Campaign files",
"path": ".",
"allow_individual": True,
"unsent_warning": False,
}
],
"allow_individual": True,
"send_without_attachments": False,
"send_without_attachments_behavior": "block",
"global": [],
"zip": {"enabled": False, "archives": []},
"missing_behavior": "warn",
"ambiguous_behavior": "ask",
},
"entries": {
"inline": [],
"defaults": {
"active": True,
"merge_to": False,
"merge_cc": True,
"merge_bcc": True,
"merge_reply_to": True,
"merge_bounce_to": True,
"merge_disposition_notification_to": True,
"combine_attachments": True,
"attachments": [],
},
},
"validation_policy": {
"missing_required_attachment": "block",
"missing_optional_attachment": "warn",
"ambiguous_attachment_match": "ask",
"unsent_attachment_files": "warn",
"ignore_empty_fields": False,
"missing_email": "block",
"template_error": "block",
},
"delivery": {
"rate_limit": {
"messages_per_minute": 5,
"concurrency": 1,
},
"imap_append_sent": {
"enabled": False,
"folder": "auto",
},
"retry": {
"max_attempts": 3,
"backoff_seconds": [60, 300, 900],
},
},
"status_tracking": {
"enabled": True,
},
}
def create_minimal_campaign(
session: Session,
*,
tenant_id: str,
user_id: str | None,
external_id: str,
name: str,
description: str | None = None,
current_flow: str = CampaignVersionFlow.CREATE.value,
current_step: str = "basics",
commit: bool = True,
) -> tuple[Campaign, CampaignVersion]:
existing = (
session.query(Campaign)
.filter(Campaign.tenant_id == tenant_id, Campaign.external_id == external_id)
.one_or_none()
)
if existing:
raise CampaignPersistenceError(
f"Campaign with id '{external_id}' already exists for this tenant"
)
campaign = Campaign(
tenant_id=tenant_id,
created_by_user_id=user_id,
owner_user_id=user_id,
external_id=external_id,
name=name,
description=description,
status=CampaignStatus.DRAFT.value,
)
session.add(campaign)
session.flush()
version = CampaignVersion(
campaign_id=campaign.id,
version_number=1,
raw_json=minimal_campaign_json(
external_id=external_id, name=name, description=description
),
schema_version="1.0",
workflow_state=CampaignVersionWorkflowState.EDITING.value,
current_flow=current_flow,
current_step=current_step,
is_complete=False,
editor_state={"created_from": "minimal_campaign"},
autosaved_at=datetime.now(UTC),
)
session.add(version)
session.flush()
campaign.current_version_id = version.id
session.add(campaign)
if commit:
session.commit()
else:
session.flush()
return campaign, version
def get_campaign_version_for_tenant(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
) -> CampaignVersion:
campaign = session.get(Campaign, campaign_id)
version = session.get(CampaignVersion, version_id)
if (
not campaign
or campaign.tenant_id != tenant_id
or not version
or version.campaign_id != campaign.id
):
raise CampaignPersistenceError("Campaign version not found")
return version
LOCKED_WORKFLOW_STATES = {
CampaignVersionWorkflowState.APPROVED.value,
CampaignVersionWorkflowState.BUILT.value,
CampaignVersionWorkflowState.QUEUED.value,
CampaignVersionWorkflowState.SENDING.value,
CampaignVersionWorkflowState.COMPLETED.value,
CampaignVersionWorkflowState.PARTIALLY_COMPLETED.value,
CampaignVersionWorkflowState.OUTCOME_UNKNOWN.value,
CampaignVersionWorkflowState.FAILED.value,
CampaignVersionWorkflowState.CANCELLED.value,
CampaignVersionWorkflowState.ARCHIVED.value,
}
def is_version_locked(version: CampaignVersion) -> bool:
"""Return True when a version is immutable and edits must fork/unlock."""
return bool(
version.locked_at
or campaign_version_user_lock_state(version)
or version.workflow_state in LOCKED_WORKFLOW_STATES
)
def ensure_current_working_version(
campaign: Campaign, version: CampaignVersion, *, action: str = "modify"
) -> None:
"""Require the campaign's single active working version.
Historical versions remain reviewable, but they never become writable in
place. Continuing from immutable history must create a new working copy,
and that copy becomes the campaign's sole current version.
"""
if campaign.current_version_id != version.id:
raise LockedCampaignVersionError(
f"Historical campaign versions are read-only and cannot be used to {action}. "
"Open the current working version instead."
)
def campaign_has_active_working_version(session: Session, campaign: Campaign) -> bool:
"""Return True while the campaign already has a non-final working version.
Validation locks and temporary user locks are still the same working
version; they must be unlocked rather than forked into parallel drafts.
"""
if not campaign.current_version_id:
return False
current = session.get(CampaignVersion, campaign.current_version_id)
if not current or current.campaign_id != campaign.id:
return False
return not is_audit_safe_version(current)
def _apply_campaign_metadata(campaign: Campaign, raw_json: dict[str, Any]) -> None:
campaign_meta = (
raw_json.get("campaign") if isinstance(raw_json.get("campaign"), dict) else {}
)
if campaign_meta:
campaign.name = campaign_meta.get("name") or campaign.name
campaign.description = campaign_meta.get("description", campaign.description)
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:
session.commit()
else:
session.flush()
def fork_campaign_version_for_edit(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
raw_json: dict[str, Any] | None = None,
current_flow: str | None = None,
current_step: str | None = None,
editor_state: dict[str, Any] | None = None,
source_filename: str | None = None,
source_base_path: str | None = None,
autosave: bool = True,
migrate_legacy_mail_settings: bool = False,
commit: bool = True,
) -> CampaignVersion:
"""Create the next sole working version from immutable campaign history.
Validation and temporary user locks are still the active working version
and must be unlocked in place. A copy is allowed only once the current
version is permanently user-locked or delivery-final.
"""
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)
_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,
migrate_legacy_mail_settings=migrate_legacy_mail_settings,
)
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,
)
_persist_forked_version(
session, campaign=campaign, version=new_version, commit=commit
)
return new_version
def lock_validated_version(
version: CampaignVersion, *, user_id: str | None = None
) -> None:
if version.locked_at is None:
version.locked_at = datetime.now(UTC)
version.locked_by_user_id = user_id
def is_version_final_locked(version: CampaignVersion) -> bool:
"""Return True when a version is part of or past delivery and must stay immutable."""
return version.workflow_state in {
CampaignVersionWorkflowState.QUEUED.value,
CampaignVersionWorkflowState.SENDING.value,
CampaignVersionWorkflowState.COMPLETED.value,
CampaignVersionWorkflowState.PARTIALLY_COMPLETED.value,
CampaignVersionWorkflowState.OUTCOME_UNKNOWN.value,
CampaignVersionWorkflowState.FAILED.value,
CampaignVersionWorkflowState.CANCELLED.value,
CampaignVersionWorkflowState.ARCHIVED.value,
}
def is_temporary_user_locked_version(version: CampaignVersion) -> bool:
return campaign_version_user_lock_state(version) == USER_LOCK_TEMPORARY
def is_permanent_user_locked_version(version: CampaignVersion) -> bool:
return campaign_version_user_lock_state(version) == USER_LOCK_PERMANENT
def is_user_locked_version(version: CampaignVersion) -> bool:
"""Return True for either reversible or permanent user-requested locks."""
return campaign_version_user_lock_state(version) is not None
def is_audit_safe_version(version: CampaignVersion) -> bool:
"""Return True when a version is immutable and cannot be unlocked."""
return is_permanent_user_locked_version(version) or is_version_final_locked(version)
def is_version_validated_and_locked(version: CampaignVersion) -> bool:
"""Return True when the version was successfully validated and locked as a review snapshot."""
validation = (
version.validation_summary
if isinstance(version.validation_summary, dict)
else {}
)
return bool(version.locked_at and validation.get("ok") is True)
def unlock_validated_campaign_version(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
commit: bool = True,
) -> CampaignVersion:
"""Unlock a validation snapshot so it can be edited again.
This is only allowed before delivery starts. Unlocking invalidates validation,
build output and queued job records for that version. Sent/final versions must
be copied instead.
"""
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="unlock")
if is_temporary_user_locked_version(version):
raise LockedCampaignVersionError(
"This version has a temporary user lock. Remove that lock before unlocking validation."
)
if is_permanent_user_locked_version(version):
raise LockedCampaignVersionError(
"This version is permanently locked and cannot be unlocked. Create an editable copy instead."
)
if is_version_final_locked(version):
raise LockedCampaignVersionError(
"This version is already queued/sent/final and cannot be unlocked. Create an editable copy instead."
)
# A version with sent jobs is final even if workflow_state was not updated for some reason.
sent_jobs = (
session.query(CampaignJob)
.filter(
CampaignJob.campaign_version_id == version.id,
CampaignJob.send_status.in_(
[JobSendStatus.SENT.value, JobSendStatus.SMTP_ACCEPTED.value]
),
)
.count()
)
if sent_jobs:
raise LockedCampaignVersionError(
"This version has sent messages and cannot be unlocked. Create an editable copy instead."
)
version.locked_at = None
version.locked_by_user_id = None
version.validation_summary = None
version.build_summary = None
clear_execution_snapshot(version)
editor_state = copy.deepcopy(version.editor_state or {})
editor_state.pop("review_send", None)
editor_state.pop("approval_gate", None)
version.editor_state = editor_state
version.workflow_state = CampaignVersionWorkflowState.EDITING.value
version.is_complete = False
session.query(CampaignIssue).filter(
CampaignIssue.campaign_version_id == version.id
).delete(synchronize_session=False)
session.query(CampaignJob).filter(
CampaignJob.campaign_version_id == version.id
).delete(synchronize_session=False)
campaign.current_version_id = version.id
campaign.status = CampaignStatus.DRAFT.value
session.add(version)
session.add(campaign)
if commit:
session.commit()
else:
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:
session.commit()
else:
session.flush()
def update_campaign_version(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
raw_json: dict[str, Any] | None = None,
current_flow: str | None = None,
current_step: str | None = None,
workflow_state: str | None = None,
is_complete: bool | None = None,
editor_state: dict[str, Any] | None = None,
source_filename: str | None = None,
source_base_path: str | None = None,
autosave: bool = False,
migrate_legacy_mail_settings: bool = False,
expected_revision: int | None = None,
commit: bool = True,
) -> CampaignVersion:
_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")
if expected_revision is not None and version.edit_revision != int(
expected_revision
):
raise RevisionConflictError(
resource_type="campaign_version",
resource_id=version.id,
current_revision=version.edit_revision,
submitted_base_revision=int(expected_revision),
refresh_path=(f"/api/v1/campaigns/{campaign.id}/versions/{version.id}"),
current_etag=strong_resource_etag(
"campaign_version",
version.id,
version.edit_revision,
),
)
if is_version_locked(version):
raise LockedCampaignVersionError(
"Campaign version is locked. Create an editable copy before changing campaign data."
)
if raw_json is not None:
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)
_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:
_invalidate_version_content(session, campaign=campaign, version=version)
try:
_persist_updated_version(
session,
campaign=campaign,
version=version,
commit=commit,
)
except StaleDataError as exc:
session.rollback()
current_revision = (
session.query(CampaignVersion.edit_revision)
.filter(CampaignVersion.id == version_id)
.scalar()
)
if current_revision is None:
raise CampaignPersistenceError(
f"Campaign version not found: {version_id}"
) from exc
raise RevisionConflictError(
resource_type="campaign_version",
resource_id=version_id,
current_revision=int(current_revision),
submitted_base_revision=int(
expected_revision
if expected_revision is not None
else version.edit_revision
),
refresh_path=(f"/api/v1/campaigns/{campaign_id}/versions/{version_id}"),
current_etag=strong_resource_etag(
"campaign_version",
version_id,
int(current_revision),
),
) from exc
return version
def update_campaign_review_state(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
inspection_complete: bool,
reviewed_message_keys: list[str],
issue_decisions: list[dict[str, Any]] | None = None,
user_id: str | None,
commit: bool = True,
) -> CampaignVersion:
"""Persist review acknowledgement without mutating the locked campaign data.
Validation locks make the campaign JSON immutable, but review metadata is
operational state attached to a specific build. It is therefore stored in
editor_state and tied to the current build token so a rebuild invalidates it.
"""
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="record review state for")
if is_version_final_locked(version):
raise LockedCampaignVersionError(
"Delivery has started; message review state can no longer be changed."
)
build_token = _campaign_review_build_token(version)
normalized_reviewed = list(
dict.fromkeys(
str(value) for value in reviewed_message_keys if str(value).strip()
)
)
normalized_decisions: list[dict[str, Any]] = []
if inspection_complete:
normalized_reviewed, normalized_decisions = _complete_campaign_review(
session,
version,
normalized_reviewed,
issue_decisions or [],
user_id=user_id,
build_token=build_token,
)
_write_campaign_review_state(
version,
build_token=build_token,
inspection_complete=inspection_complete,
reviewed_message_keys=normalized_reviewed,
issue_decisions=normalized_decisions,
user_id=user_id,
)
session.add(version)
if commit:
session.commit()
else:
session.flush()
return version
def _campaign_review_build_token(version: CampaignVersion) -> str:
build_summary = (
version.build_summary if isinstance(version.build_summary, dict) else {}
)
if not build_summary:
raise CampaignPersistenceError("Build messages before recording review state.")
build_token = str(
build_summary.get("build_token") or build_summary.get("built_at") or ""
).strip()
if build_token:
return build_token
build_token = uuid4().hex
updated_summary = copy.deepcopy(build_summary)
updated_summary["build_token"] = build_token
version.build_summary = updated_summary
return build_token
def _complete_campaign_review(
session: Session,
version: CampaignVersion,
reviewed_message_keys: list[str],
issue_decisions: list[dict[str, Any]],
*,
user_id: str | None,
build_token: str,
) -> tuple[list[str], list[dict[str, Any]]]:
jobs = (
session.query(CampaignJob)
.filter(CampaignJob.campaign_version_id == version.id)
.order_by(CampaignJob.entry_index.asc())
.all()
)
blocking = [
job
for job in jobs
if job.build_status != "built" or job.validation_status == "blocked"
]
if blocking:
raise CampaignPersistenceError(
"Blocked or failed messages must be resolved before review can be completed."
)
missing = sorted(_required_review_keys(jobs) - set(reviewed_message_keys))
if missing:
raise CampaignPersistenceError(
"Messages requiring an explicit decision must be opened before review can be completed: "
+ ", ".join(missing)
)
decisions = _normalize_review_issue_decisions(
jobs,
issue_decisions,
user_id=user_id,
build_token=build_token,
)
return (
list(
dict.fromkeys([*reviewed_message_keys, *_bulk_acceptable_review_keys(jobs)])
),
decisions,
)
def _normalize_review_issue_decisions(
jobs: list[CampaignJob],
requested: list[dict[str, Any]],
*,
user_id: str | None,
build_token: str,
decided_at: datetime | None = None,
) -> list[dict[str, Any]]:
jobs_by_id = {job.id: job for job in jobs}
requested_by_job: dict[str, dict[str, Any]] = {}
for item in requested:
job_id = str(item.get("job_id") or "").strip()
if not job_id or job_id not in jobs_by_id:
raise CampaignPersistenceError(
"A review decision references a message outside the current build."
)
if jobs_by_id[job_id].validation_status != "needs_review":
raise CampaignPersistenceError(
"Review decisions are accepted only for messages requiring review."
)
if job_id in requested_by_job:
raise CampaignPersistenceError(
"Only one review decision may be recorded per built message."
)
if str(item.get("decision") or "accept") != "accept":
raise CampaignPersistenceError("Unsupported campaign review decision.")
requested_by_job[job_id] = item
timestamp = (decided_at or datetime.now(UTC)).isoformat()
normalized: list[dict[str, Any]] = []
for job in jobs:
reviewable_issues = [
issue
for issue in (job.issues_snapshot or [])
if isinstance(issue, dict)
and str(issue.get("behavior") or "").lower() == "ask"
]
attachment_overrides = [
issue
for issue in reviewable_issues
if str(issue.get("source") or "").startswith("attachments")
]
decision = requested_by_job.get(job.id)
if attachment_overrides and decision is None:
raise CampaignPersistenceError(
"Attachment exceptions require an explicit reason before review can be completed "
f"for message {job.entry_id or job.entry_index}."
)
if decision is None:
continue
reason = str(decision.get("reason") or "").strip()
if attachment_overrides and not reason:
raise CampaignPersistenceError(
"Attachment exception decisions require a reason."
)
evidence_issues = reviewable_issues or [
issue for issue in (job.issues_snapshot or []) if isinstance(issue, dict)
]
issue_payload = [
{
"code": str(issue.get("code") or ""),
"behavior": str(issue.get("behavior") or ""),
"source": str(issue.get("source") or ""),
"details": issue.get("details")
if isinstance(issue.get("details"), dict)
else {},
}
for issue in evidence_issues
]
fingerprint = hashlib.sha256(
json.dumps(
issue_payload,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
normalized.append(
{
"job_id": job.id,
"review_key": str(job.entry_id or job.entry_index),
"decision": "accept",
"reason": reason or None,
"actor_user_id": user_id,
"decided_at": timestamp,
"build_token": build_token,
"message_sha256": job.eml_sha256,
"issue_fingerprint": fingerprint,
"issue_codes": sorted(
{
str(issue.get("code") or "")
for issue in evidence_issues
if issue.get("code")
}
),
}
)
return normalized
def _required_review_keys(jobs: list[CampaignJob]) -> set[str]:
return {
str(job.entry_id or job.entry_index)
for job in jobs
if job.validation_status == "needs_review"
}
def _bulk_acceptable_review_keys(jobs: list[CampaignJob]) -> list[str]:
return [
str(job.entry_id or job.entry_index)
for job in jobs
if job.validation_status in {"warning", "excluded"}
]
def _write_campaign_review_state(
version: CampaignVersion,
*,
build_token: str,
inspection_complete: bool,
reviewed_message_keys: list[str],
issue_decisions: list[dict[str, Any]],
user_id: str | None,
) -> None:
editor_state = copy.deepcopy(version.editor_state or {})
editor_state["review_send"] = {
"build_token": build_token,
"inspection_complete": bool(inspection_complete),
"reviewed_message_keys": reviewed_message_keys,
"issue_decisions": issue_decisions,
"updated_at": datetime.now(UTC).isoformat(),
"updated_by_user_id": user_id,
}
version.editor_state = editor_state
def lock_campaign_version_temporarily(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
user_id: str | None,
commit: bool = True,
) -> CampaignVersion:
"""Apply a reversible user-requested lock without changing workflow state."""
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="lock")
if is_version_final_locked(version):
raise LockedCampaignVersionError(
"Delivery/final versions are permanently locked and cannot receive a temporary user lock."
)
if is_permanent_user_locked_version(version):
raise LockedCampaignVersionError("This version is already permanently locked.")
if is_temporary_user_locked_version(version):
return version
if version.locked_at:
raise LockedCampaignVersionError(
"This version is already temporarily locked by validation. Unlock validation before applying a user lock."
)
version.user_lock_state = USER_LOCK_TEMPORARY
version.user_locked_at = datetime.now(UTC)
version.user_locked_by_user_id = user_id
session.add(version)
if commit:
session.commit()
else:
session.flush()
return version
def unlock_user_locked_campaign_version(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
commit: bool = True,
) -> CampaignVersion:
"""Remove a reversible user lock without invalidating campaign data."""
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="unlock")
state = campaign_version_user_lock_state(version)
if state == USER_LOCK_PERMANENT:
raise LockedCampaignVersionError(
"Permanently locked versions cannot be unlocked. Create an editable copy instead."
)
if state != USER_LOCK_TEMPORARY:
raise LockedCampaignVersionError(
"This version does not have a temporary user lock."
)
if is_version_final_locked(version):
raise LockedCampaignVersionError(
"Delivery/final versions cannot be unlocked. Create an editable copy instead."
)
version.user_lock_state = None
version.user_locked_at = None
version.user_locked_by_user_id = None
session.add(version)
if commit:
session.commit()
else:
session.flush()
return version
def permanently_lock_campaign_version(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
user_id: str | None,
commit: bool = True,
) -> CampaignVersion:
"""Apply an irreversible user lock.
The version remains in its current workflow state so the campaign itself is
not silently archived. Future changes must be made in an editable copy.
"""
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="lock permanently")
if is_version_final_locked(version):
raise LockedCampaignVersionError(
"This version is already permanently locked by its delivery/final state."
)
if is_permanent_user_locked_version(version):
return version
now = datetime.now(UTC)
version.user_lock_state = USER_LOCK_PERMANENT
version.user_locked_at = now
version.user_locked_by_user_id = user_id
# Retain published_at as a compatibility marker for existing integrations.
version.published_at = version.published_at or now
session.add(version)
if commit:
session.commit()
else:
session.flush()
return version
def publish_campaign_version(
session: Session,
*,
tenant_id: str,
campaign_id: str,
version_id: str,
user_id: str | None = None,
commit: bool = True,
) -> CampaignVersion:
"""Backwards-compatible alias for the permanent user lock."""
return permanently_lock_campaign_version(
session,
tenant_id=tenant_id,
campaign_id=campaign_id,
version_id=version_id,
user_id=user_id,
commit=commit,
)
def validate_campaign_partial(
raw_json: dict[str, Any], *, section: str | None = None
) -> dict[str, Any]:
"""Lightweight UI-facing validation for incomplete campaign working copies.
This is intentionally less strict than campaign.schema.json validation. It
lets the WebUI autosave and validate one wizard step at a time.
"""
collector = _PartialValidationCollector(section=section)
_validate_partial_basics(collector, _dict_value(raw_json, "campaign"))
recipients = _dict_value(raw_json, "recipients")
_validate_partial_sender(collector, recipients)
_validate_partial_mail_profile(collector, raw_json)
_validate_partial_recipients(collector, _dict_value(raw_json, "entries"))
_validate_partial_template(collector, _dict_value(raw_json, "template"))
_validate_partial_attachments(collector, _dict_value(raw_json, "attachments"))
_validate_partial_delivery(collector, _dict_value(raw_json, "delivery"))
return collector.result()
def _dict_value(value: dict[str, Any], key: str) -> dict[str, Any]:
candidate = value.get(key)
return candidate if isinstance(candidate, dict) else {}
def _validate_partial_basics(
collector: _PartialValidationCollector, campaign: dict[str, Any]
) -> None:
if not campaign.get("id"):
collector.issue(
"error",
"basics",
"campaign.id",
"missing_campaign_id",
"Campaign id is required.",
)
if not campaign.get("name"):
collector.issue(
"error",
"basics",
"campaign.name",
"missing_campaign_name",
"Campaign name is required.",
)
def _validate_partial_sender(
collector: _PartialValidationCollector, recipients: dict[str, Any]
) -> None:
sender = recipients.get("from") if isinstance(recipients.get("from"), dict) else {}
if not sender.get("email"):
collector.issue(
"warning",
"sender",
"recipients.from.email",
"missing_sender_email",
"Sender email is not configured yet.",
)
def _validate_partial_mail_profile(
collector: _PartialValidationCollector, raw_json: dict[str, Any]
) -> None:
violations = campaign_mail_profile_boundary_violations(raw_json)
if violations:
collector.issue(
"error",
"sender",
"server",
"campaign_local_mail_transport_forbidden",
"Campaign-local SMTP/IMAP settings are not supported. Select or migrate to an authorized Mail-module profile.",
)
elif campaign_mail_profile_id(raw_json) is None:
collector.issue(
"warning",
"sender",
"server.mail_profile_id",
"missing_mail_profile",
"Select an authorized Mail-module profile before validating or delivering this campaign.",
)
def _validate_partial_recipients(
collector: _PartialValidationCollector, entries: dict[str, Any]
) -> None:
has_inline = bool(entries.get("inline"))
has_source = isinstance(entries.get("source"), dict)
if not has_inline and not has_source:
collector.issue(
"warning",
"recipients",
"entries",
"missing_recipients",
"No inline recipients or external recipient source configured yet.",
)
if has_source and not _entries_source_has_email_mapping(entries):
collector.issue(
"warning",
"recipients",
"entries.mapping",
"missing_email_mapping",
"No email field mapping is configured.",
)
def _entries_source_has_email_mapping(entries: dict[str, Any]) -> bool:
mapping = entries.get("mapping") if isinstance(entries.get("mapping"), dict) else {}
return any(key in mapping for key in ("to.0.email", "to.email", "email"))
def _validate_partial_template(
collector: _PartialValidationCollector, template: dict[str, Any]
) -> None:
source_template = (
template.get("source") if isinstance(template.get("source"), dict) else {}
)
if not template.get("subject") and not source_template.get("subject_path"):
collector.issue(
"warning",
"template",
"template.subject",
"missing_subject",
"Template subject is empty.",
)
body_state = _partial_template_body_state(template, source_template)
_validate_partial_template_body(collector, body_state)
def _partial_template_body_state(
template: dict[str, Any], source_template: dict[str, Any]
) -> dict[str, Any]:
return {
"mode": template.get("body_mode")
if template.get("body_mode") in {"text", "html", "both"}
else None,
"has_text": bool(template.get("text"))
or bool(source_template.get("text_path")),
"has_html": bool(template.get("html"))
or bool(source_template.get("html_path")),
"has_source": bool(source_template),
}
def _validate_partial_template_body(
collector: _PartialValidationCollector, state: dict[str, Any]
) -> None:
mode = state["mode"]
has_text = bool(state["has_text"])
has_html = bool(state["has_html"])
if mode == "text" and not has_text:
collector.issue(
"warning",
"template",
"template.text",
"missing_template_text_body",
"Template body mode is text only, but no text body is configured.",
)
elif mode == "html" and not has_html:
collector.issue(
"warning",
"template",
"template.html",
"missing_template_html_body",
"Template body mode is HTML only, but no HTML body is configured.",
)
elif mode == "both":
_validate_partial_dual_body_template(
collector, has_text=has_text, has_html=has_html
)
elif not has_text and not has_html and not state["has_source"]:
collector.issue(
"warning",
"template",
"template",
"missing_template_body",
"No text, HTML or file-based template body configured yet.",
)
def _validate_partial_dual_body_template(
collector: _PartialValidationCollector, *, has_text: bool, has_html: bool
) -> None:
if not has_text:
collector.issue(
"warning",
"template",
"template.text",
"missing_template_text_body",
"Template body mode is both, but no text body is configured.",
)
if not has_html:
collector.issue(
"warning",
"template",
"template.html",
"missing_template_html_body",
"Template body mode is both, but no HTML body is configured.",
)
def _validate_partial_attachments(
collector: _PartialValidationCollector, attachments: dict[str, Any]
) -> None:
base_paths = (
attachments.get("base_paths")
if isinstance(attachments.get("base_paths"), list)
else []
)
has_named_base_path = any(
isinstance(item, dict) and item.get("path") for item in base_paths
)
if not has_named_base_path and not attachments.get("base_path"):
collector.issue(
"info",
"attachments",
"attachments.base_path",
"missing_attachment_base_path",
"Attachment base path is not configured yet.",
)
def _validate_partial_delivery(
collector: _PartialValidationCollector, delivery: dict[str, Any]
) -> None:
rate_limit = (
delivery.get("rate_limit")
if isinstance(delivery.get("rate_limit"), dict)
else {}
)
messages_per_minute = rate_limit.get("messages_per_minute")
if messages_per_minute is None:
return
try:
if int(messages_per_minute) < 1:
collector.issue(
"error",
"send",
"delivery.rate_limit.messages_per_minute",
"invalid_rate_limit",
"Messages per minute must be at least 1.",
)
except (TypeError, ValueError):
collector.issue(
"error",
"send",
"delivery.rate_limit.messages_per_minute",
"invalid_rate_limit",
"Messages per minute must be a number.",
)