govoplan-campaign/src/govoplan_campaign/backend/campaign/lifecycle.py

from __future__ import annotations

import hashlib
import hmac
import json
from dataclasses import dataclass
from datetime import datetime
from typing import Any

from sqlalchemy.orm import Session

from govoplan_campaign.backend.db.models import (
    Campaign,
    CampaignJob,
    CampaignShare,
    CampaignVersion,
)
from govoplan_core.auth import ApiPrincipal, has_scope


POLICY_ID = "campaign.lifecycle"
POLICY_VERSION = "1"

_ACTIVE_QUEUE_STATES = {"queued", "sending"}
_ACTIVE_SEND_STATES = {"queued", "claimed", "sending", "outcome_unknown"}
_ACTIVE_POSTBOX_STATES = {"pending", "delivering", "outcome_unknown"}
_ACTIVE_PRINT_STATES = {"ready", "accepting"}
_ACTIVE_IMAP_STATES = {"pending", "appending", "outcome_unknown"}


@dataclass(frozen=True)
class LifecycleDecision:
    allowed: bool
    reason: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {"allowed": self.allowed, "reason": self.reason}


def _timestamp(value: datetime | None) -> str | None:
    return value.isoformat() if value is not None else None


def _canonical_hash(value: object) -> str:
    return hashlib.sha256(
        json.dumps(
            value,
            ensure_ascii=True,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    ).hexdigest()


def _protected_version(version: CampaignVersion) -> bool:
    return any(
        value is not None
        for value in (
            version.locked_at,
            version.user_lock_state,
            version.published_at,
            version.execution_snapshot_at,
        )
    )


def _active_delivery(job: CampaignJob) -> bool:
    return any(
        (
            job.queue_status in _ACTIVE_QUEUE_STATES,
            job.send_status in _ACTIVE_SEND_STATES,
            job.postbox_status in _ACTIVE_POSTBOX_STATES,
            job.print_status in _ACTIVE_PRINT_STATES,
            job.imap_status in _ACTIVE_IMAP_STATES,
        )
    )


def campaign_lifecycle_policy(
    session: Session,
    *,
    campaign: Campaign,
    principal: ApiPrincipal,
    version_id: str | None = None,
) -> dict[str, Any]:
    versions = (
        session.query(CampaignVersion)
        .filter(CampaignVersion.campaign_id == campaign.id)
        .order_by(CampaignVersion.version_number.asc())
        .all()
    )
    jobs = (
        session.query(CampaignJob)
        .filter(CampaignJob.campaign_id == campaign.id)
        .order_by(CampaignJob.id.asc())
        .all()
    )
    shares = (
        session.query(CampaignShare)
        .filter(
            CampaignShare.campaign_id == campaign.id,
            CampaignShare.revoked_at.is_(None),
        )
        .order_by(CampaignShare.id.asc())
        .all()
    )
    selected_version = next(
        (version for version in versions if version.id == version_id),
        None,
    )

    snapshot = {
        "policy_id": POLICY_ID,
        "policy_version": POLICY_VERSION,
        "campaign": {
            "id": campaign.id,
            "status": campaign.status,
            "current_version_id": campaign.current_version_id,
            "updated_at": _timestamp(campaign.updated_at),
        },
        "versions": [
            {
                "id": version.id,
                "version_number": version.version_number,
                "edit_revision": version.edit_revision,
                "workflow_state": version.workflow_state,
                "locked_at": _timestamp(version.locked_at),
                "user_lock_state": version.user_lock_state,
                "published_at": _timestamp(version.published_at),
                "execution_snapshot_at": _timestamp(version.execution_snapshot_at),
                "archived_at": _timestamp(version.archived_at),
                "updated_at": _timestamp(version.updated_at),
            }
            for version in versions
        ],
        "jobs": [
            {
                "id": job.id,
                "queue_status": job.queue_status,
                "send_status": job.send_status,
                "postbox_status": job.postbox_status,
                "print_status": job.print_status,
                "imap_status": job.imap_status,
                "updated_at": _timestamp(job.updated_at),
            }
            for job in jobs
        ],
        "active_share_ids": [share.id for share in shares],
        "selected_version_id": version_id,
    }
    token = _canonical_hash(snapshot)

    active_delivery = any(_active_delivery(job) for job in jobs)
    protected_versions = any(_protected_version(version) for version in versions)

    archive = LifecycleDecision(True)
    if not has_scope(principal, "campaigns:campaign:archive"):
        archive = LifecycleDecision(False, "Missing campaign archive permission.")
    elif campaign.status in {"archived", "deleted"}:
        archive = LifecycleDecision(False, "The campaign is already archived or deleted.")
    elif active_delivery:
        archive = LifecycleDecision(
            False,
            "Active or uncertain delivery must be resolved before archiving.",
        )

    delete = LifecycleDecision(True)
    if not has_scope(principal, "campaigns:campaign:delete"):
        delete = LifecycleDecision(False, "Missing campaign delete permission.")
    elif campaign.status != "draft":
        delete = LifecycleDecision(False, "Only untouched draft campaigns can be deleted.")
    elif jobs:
        delete = LifecycleDecision(
            False,
            "Campaigns with built or delivery jobs must be archived instead of deleted.",
        )
    elif protected_versions:
        delete = LifecycleDecision(
            False,
            "Audit-relevant campaign versions must be archived instead of deleted.",
        )
    elif shares:
        delete = LifecycleDecision(
            False,
            "Revoke active campaign shares before deleting the untouched draft.",
        )

    copy = LifecycleDecision(True)
    if not has_scope(principal, "campaigns:campaign:copy"):
        copy = LifecycleDecision(False, "Missing campaign copy permission.")
    elif not has_scope(principal, "campaigns:recipient:read"):
        copy = LifecycleDecision(False, "Recipient read permission is required to copy a campaign.")
    elif version_id is not None and selected_version is None:
        copy = LifecycleDecision(False, "The selected source version does not exist.")

    archive_version = LifecycleDecision(True)
    if not has_scope(principal, "campaigns:campaign:archive"):
        archive_version = LifecycleDecision(False, "Missing campaign archive permission.")
    elif version_id is None or selected_version is None:
        archive_version = LifecycleDecision(False, "Select a historical campaign version.")
    elif selected_version.id == campaign.current_version_id:
        archive_version = LifecycleDecision(False, "The current campaign version cannot be archived.")
    elif selected_version.archived_at is not None:
        archive_version = LifecycleDecision(False, "The historical version is already archived.")

    return {
        "policy_id": POLICY_ID,
        "policy_version": POLICY_VERSION,
        "state_token": token,
        "actions": {
            "archive_campaign": archive.as_dict(),
            "delete_campaign": delete.as_dict(),
            "copy_campaign": copy.as_dict(),
            "archive_version": archive_version.as_dict(),
        },
        "provenance": {
            "source": "built_in",
            "rules": (
                "permission",
                "campaign_state",
                "retained_evidence",
                "active_delivery",
                "optimistic_concurrency",
            ),
            "evidence_retention": "Versions, delivery outcomes, reports, and audit records are never deleted by archival.",
        },
    }


def assert_lifecycle_state_token(actual: str, expected: str) -> None:
    if not hmac.compare_digest(actual, expected):
        raise ValueError(
            "Campaign state changed after this action was prepared. Reload and review the lifecycle decision again."
        )