from __future__ import annotations from datetime import datetime from typing import Any from sqlalchemy import event from sqlalchemy.orm import Session as OrmSession from govoplan_core.core.change_sequence import record_change from govoplan_core.core.sqlalchemy_change_tracking import ( ensure_object_id, has_attr_changes, object_state, operation_for_object, previous_value, ) from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, CampaignJob, CampaignShare, CampaignVersion, ImapAppendAttempt, SendAttempt, new_uuid, ) CAMPAIGNS_MODULE_ID = "campaigns" CAMPAIGNS_COLLECTION = "campaigns.campaigns" CAMPAIGN_VERSIONS_COLLECTION = "campaigns.versions" CAMPAIGN_JOBS_COLLECTION = "campaigns.jobs" CAMPAIGN_ISSUES_COLLECTION = "campaigns.issues" CAMPAIGN_ATTEMPTS_COLLECTION = "campaigns.attempts" _REGISTERED = False def register_campaign_change_tracking() -> None: global _REGISTERED if _REGISTERED: return event.listen(OrmSession, "before_flush", _record_campaign_changes) _REGISTERED = True def _record_campaign_changes(session: OrmSession, _flush_context: object, _instances: object) -> None: for obj in tuple(session.new) + tuple(session.dirty) + tuple(session.deleted): if isinstance(obj, Campaign): _record_campaign_change(session, obj) elif isinstance(obj, CampaignShare): _record_share_visibility_change(session, obj) elif isinstance(obj, CampaignVersion): _record_version_change(session, obj) elif isinstance(obj, CampaignJob): _record_job_change(session, obj) elif isinstance(obj, CampaignIssue): _record_issue_change(session, obj) elif isinstance(obj, (SendAttempt, ImapAppendAttempt)): _record_attempt_change(session, obj) def _record_campaign_change(session: OrmSession, campaign: Campaign) -> None: operation = _operation_for_campaign( campaign, changed_attrs=( "external_id", "name", "description", "status", "current_version_id", "owner_user_id", "owner_group_id", "settings", "mail_profile_policy", ), ) if operation is None: return campaign_id = _ensure_id(campaign) record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGNS_COLLECTION, resource_type="campaign", resource_id=campaign_id, operation=operation, tenant_id=campaign.tenant_id, actor_type="user" if campaign.created_by_user_id else None, actor_id=campaign.created_by_user_id, payload=_campaign_payload(campaign), ) def _record_share_visibility_change(session: OrmSession, share: CampaignShare) -> None: state = object_state(share) if not share.campaign_id: return if not state.deleted and not state.pending and not has_attr_changes( state, ("campaign_id", "target_type", "target_id", "permission", "revoked_at"), ): return _ensure_id(share) operation = "deleted" if state.deleted or share.revoked_at is not None else "updated" campaign = session.get(Campaign, share.campaign_id) payload = _campaign_payload(campaign) if campaign is not None else {"campaign_id": share.campaign_id} payload.update({ "share_id": share.id, "share_target_type": share.target_type, "share_target_id": share.target_id, "share_permission": share.permission, "share_revoked_at": _isoformat(share.revoked_at), }) record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGNS_COLLECTION, resource_type="campaign", resource_id=share.campaign_id, operation=operation, tenant_id=share.tenant_id, actor_type="user" if share.created_by_user_id else None, actor_id=share.created_by_user_id, payload=payload, ) def _record_version_change(session: OrmSession, version: CampaignVersion) -> None: operation = _operation_for_object( version, changed_attrs=( "raw_json", "schema_version", "source_filename", "source_base_path", "workflow_state", "current_flow", "current_step", "is_complete", "editor_state", "autosaved_at", "published_at", "locked_at", "locked_by_user_id", "user_lock_state", "user_locked_at", "user_locked_by_user_id", "validation_summary", "build_summary", "execution_snapshot_hash", "execution_snapshot_at", ), ) if operation is None: return version_id = _ensure_id(version) campaign = session.get(Campaign, version.campaign_id) if version.campaign_id else None record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGN_VERSIONS_COLLECTION, resource_type="campaign_version", resource_id=version_id, operation=operation, tenant_id=campaign.tenant_id if campaign is not None else None, payload={ **(_campaign_payload(campaign) if campaign is not None else {"campaign_id": version.campaign_id}), "version_id": version_id, "version_number": version.version_number, "workflow_state": version.workflow_state, "current_flow": version.current_flow, "current_step": version.current_step, }, ) def _record_job_change(session: OrmSession, job: CampaignJob) -> None: operation = _operation_for_object( job, changed_attrs=( "build_status", "validation_status", "queue_status", "send_status", "imap_status", "attempt_count", "last_error", "queued_at", "claimed_at", "smtp_started_at", "outcome_unknown_at", "sent_at", "resolved_recipients", "resolved_attachments", "issues_snapshot", ), ) if operation is None: return job_id = _ensure_id(job) campaign = session.get(Campaign, job.campaign_id) if job.campaign_id else None record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGN_JOBS_COLLECTION, resource_type="campaign_job", resource_id=job_id, operation=operation, tenant_id=job.tenant_id, payload={ **(_campaign_payload(campaign) if campaign is not None else {"campaign_id": job.campaign_id}), "job_id": job_id, "version_id": job.campaign_version_id, "entry_index": job.entry_index, "entry_id": job.entry_id, "build_status": job.build_status, "validation_status": job.validation_status, "queue_status": job.queue_status, "send_status": job.send_status, "imap_status": job.imap_status, }, ) def _record_issue_change(session: OrmSession, issue: CampaignIssue) -> None: operation = _operation_for_object(issue, changed_attrs=("severity", "code", "message", "source", "behavior")) if operation is None: return issue_id = _ensure_id(issue) campaign = session.get(Campaign, issue.campaign_id) if issue.campaign_id else None record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGN_ISSUES_COLLECTION, resource_type="campaign_issue", resource_id=issue_id, operation=operation, tenant_id=issue.tenant_id, payload={ **(_campaign_payload(campaign) if campaign is not None else {"campaign_id": issue.campaign_id}), "issue_id": issue_id, "version_id": issue.campaign_version_id, "job_id": issue.job_id, "severity": issue.severity, "code": issue.code, }, ) def _record_attempt_change(session: OrmSession, attempt: SendAttempt | ImapAppendAttempt) -> None: operation = _operation_for_object( attempt, changed_attrs=("status", "claim_token", "smtp_status_code", "smtp_response", "error_type", "error_message", "folder"), ) if operation is None: return attempt_id = _ensure_id(attempt) job = session.get(CampaignJob, attempt.job_id) if attempt.job_id else None if job is None: return campaign = session.get(Campaign, job.campaign_id) if job.campaign_id else None record_change( session, module_id=CAMPAIGNS_MODULE_ID, collection=CAMPAIGN_ATTEMPTS_COLLECTION, resource_type="campaign_attempt", resource_id=attempt_id, operation=operation, tenant_id=job.tenant_id, payload={ **(_campaign_payload(campaign) if campaign is not None else {"campaign_id": job.campaign_id}), "attempt_id": attempt_id, "attempt_kind": "imap" if isinstance(attempt, ImapAppendAttempt) else "smtp", "job_id": job.id, "version_id": job.campaign_version_id, }, ) def _operation_for_campaign(obj: Campaign, *, changed_attrs: tuple[str, ...]) -> str | None: state = object_state(obj) if state.pending: return "created" if state.deleted: return "deleted" if not has_attr_changes(state, changed_attrs): return None status_history = state.attrs.status.history if status_history.has_changes() and any(value == "deleted" for value in status_history.added): return "deleted" return "updated" def _operation_for_object(obj: object, *, changed_attrs: tuple[str, ...]) -> str | None: return operation_for_object(obj, changed_attrs=changed_attrs) def _ensure_id(obj: object) -> str: return ensure_object_id(obj, new_uuid) def _campaign_payload(campaign: Campaign | None) -> dict[str, Any]: if campaign is None: return {} return { "campaign_id": campaign.id, "owner_user_id": campaign.owner_user_id, "owner_group_id": campaign.owner_group_id, "previous_owner_user_id": previous_value(campaign, "owner_user_id"), "previous_owner_group_id": previous_value(campaign, "owner_group_id"), "status": campaign.status, "previous_status": previous_value(campaign, "status"), "current_version_id": campaign.current_version_id, "previous_current_version_id": previous_value(campaign, "current_version_id"), } def _isoformat(value: datetime | None) -> str | None: return value.isoformat() if value else None __all__ = [ "CAMPAIGNS_COLLECTION", "CAMPAIGNS_MODULE_ID", "CAMPAIGN_ATTEMPTS_COLLECTION", "CAMPAIGN_ISSUES_COLLECTION", "CAMPAIGN_JOBS_COLLECTION", "CAMPAIGN_VERSIONS_COLLECTION", "register_campaign_change_tracking", ]