from __future__ import annotations from collections import Counter from collections.abc import Mapping, Sequence from datetime import datetime, timezone from sqlalchemy import Text, and_, cast, func, or_ from sqlalchemy.orm import Session from govoplan_campaign.backend.db.models import ( AttachmentInstance, Campaign, CampaignCollaborationEntry, CampaignIssue, CampaignJob, CampaignMessageAction, CampaignMessageActionAttempt, CampaignSchedule, CampaignShare, CampaignVersion, CampaignWorkAssignment, CampaignWorkAssignmentEvent, ImapAppendAttempt, PostboxDeliveryAttempt, PrintOutputAttempt, RecipientImportMappingProfile, SendAttempt, ) from govoplan_core.core.dsar import ( DsarErasureActionRef, DsarExecutionResultRef, DsarRecordRef, DsarSubjectRef, dsar_capability_name, ) CAMPAIGN_DSAR_CAPABILITY = dsar_capability_name("campaigns") _MAX_RECORDS = 5_000 _SECRET_PARTS = ( "password", "secret", "token", "credential", "authorization", "private_key", "claim_token", "idempotency_key", ) _CONTENT_KEYS = { "attachments", "body", "content", "html", "template", "text", } class CampaignDsarProvider: provider_id = "campaigns" module_id = "campaigns" def search_subject( self, session: object, *, tenant_id: str, subject: DsarSubjectRef, ) -> Sequence[DsarRecordRef]: db = _session(session) subject_user_id = _subject_user_id(subject) subject_email = _subject_email(subject) references = _campaign_references(subject) if subject_user_id is None and subject_email is None and not references: return () records: list[DsarRecordRef] = [] def append(record: DsarRecordRef) -> None: if len(records) >= _MAX_RECORDS: raise ValueError( "Campaign DSAR match limit exceeded; narrow the subject selectors." ) records.append(record) jobs = _matching_jobs( db, tenant_id=tenant_id, email=subject_email, job_id=references.get("job"), entry_id=references.get("entry"), ) job_ids = {row.id for row in jobs} job_version_ids = {row.campaign_version_id for row in jobs} campaign_ids = {row.campaign_id for row in jobs} evidence_campaign_ids = set(campaign_ids) versions = _matching_versions( db, tenant_id=tenant_id, subject_user_id=subject_user_id, email=subject_email, job_version_ids=job_version_ids, version_id=references.get("version"), ) version_ids = {row.id for row in versions} campaign_ids.update(row.campaign_id for row in versions) if references.get("campaign"): campaign_ids.add(references["campaign"]) campaigns = _matching_campaigns( db, tenant_id=tenant_id, subject_user_id=subject_user_id, campaign_ids=campaign_ids, ) campaign_ids = {row.id for row in campaigns} for campaign in campaigns: immutable = campaign.status != "draft" append( _record( "campaign", campaign.id, "campaign_workspace", campaign.name, { "match_fields": _matching_fields( campaign, subject_user_id, ("created_by_user_id", "owner_user_id"), ), "recipient_context": campaign.id in {job.campaign_id for job in jobs}, "external_id": campaign.external_id, "name": campaign.name, "description": campaign.description, "status": campaign.status, }, observed_at=campaign.updated_at, immutable=immutable, retention_reason=( "Non-draft Campaign state is institutional workflow and delivery evidence." if immutable else None ), source_path=f"/campaigns/{campaign.id}", ) ) jobs_by_version: Counter[str] = Counter(row.campaign_version_id for row in jobs) for version in versions: fragments = _matching_entry_fragments(version.raw_json, subject_email) immutable = _version_is_evidence(version, jobs_by_version[version.id]) match_fields = _matching_fields( version, subject_user_id, ( "locked_by_user_id", "user_locked_by_user_id", "archived_by_user_id", ), ) if fragments: match_fields.insert(0, "raw_json.recipient") if version.id in job_version_ids: match_fields.insert(0, "campaign_job.recipient") append( _record( "campaign_version", version.id, "campaign_recipient_snapshot", f"Campaign version {version.version_number}", { "match_fields": match_fields, "campaign_id": version.campaign_id, "version_number": version.version_number, "workflow_state": version.workflow_state, "published_at": _iso(version.published_at), "locked_at": _iso(version.locked_at), "archived_at": _iso(version.archived_at), "matching_entries": fragments, }, observed_at=version.updated_at, immutable=immutable, retention_reason=( "Built, locked, published, or terminal Campaign versions are retained as execution evidence and are redacted only by Campaign retention policy." if immutable else None ), source_path=f"/campaigns/{version.campaign_id}", ) ) immutable_jobs: dict[str, bool] = {} for job in jobs: immutable = _job_is_evidence(job) immutable_jobs[job.id] = immutable append( _record( "campaign_recipient_job", job.id, "campaign_recipient_delivery", job.recipient_email or f"Recipient {job.entry_index}", { "campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id, "entry_index": job.entry_index, "entry_id": job.entry_id, "recipient_email": job.recipient_email, "subject": job.subject, "build_status": job.build_status, "validation_status": job.validation_status, "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, "attempt_count": job.attempt_count, "message_sha256": job.eml_sha256, "message_size_bytes": job.eml_size_bytes, "queued_at": _iso(job.queued_at), "sent_at": _iso(job.sent_at), "matched_recipients": _matching_recipient_fragment( job.resolved_recipients, subject_email, ), }, observed_at=job.updated_at, immutable=immutable, retention_reason=( "Built message and delivery state is retained under Campaign delivery-evidence policy." if immutable else None ), source_path=f"/campaigns/{job.campaign_id}/report", ) ) if ( job.eml_sha256 or job.eml_size_bytes or job.eml_storage_key or job.eml_local_path ): append( _record( "campaign_message_artifact", job.id, "generated_message_artifact", f"Generated message for job {job.id}", { "campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id, "job_id": job.id, "sha256": job.eml_sha256, "size_bytes": job.eml_size_bytes, "artifact_present": bool( job.eml_storage_key or job.eml_local_path ), }, observed_at=job.updated_at, immutable=True, retention_reason=( "Generated message bytes and their locator follow Campaign artifact retention and recovery controls." ), source_path=f"/campaigns/{job.campaign_id}/report", ) ) self._append_delivery_evidence( db, append=append, tenant_id=tenant_id, job_ids=job_ids, immutable_jobs=immutable_jobs, ) self._append_report_projections(append=append, jobs=jobs) self._append_member_resources( db, append=append, tenant_id=tenant_id, subject_user_id=subject_user_id, subject_account_id=subject.account_id, campaign_ids=evidence_campaign_ids, version_ids=version_ids, ) return tuple(records) def plan_erasure( self, session: object, *, tenant_id: str, subject: DsarSubjectRef, records: Sequence[DsarRecordRef], ) -> Sequence[DsarErasureActionRef]: del session subject_user_id = _subject_user_id(subject) actions: list[DsarErasureActionRef] = [] for record in records: if ( record.provider_id != self.provider_id or record.module_id != self.module_id ): raise ValueError("Campaign DSAR received a foreign provider record.") match_fields = {str(value) for value in record.data.get("match_fields", ())} if record.immutable_evidence: actions.append( _action( f"campaigns:retain:{record.resource_type}:{record.resource_id}", "retain", record, f"Retain {record.title}", record.retention_reason or "Institutional Campaign evidence must be retained.", executable=False, ) ) else: actions.append( _action( f"campaigns:review:{record.resource_type}:{record.resource_id}", "manual_review", record, f"Review {record.title}", ( "Recipient data can span a Campaign version, generated message, " "attachments, and delivery projections. Apply Campaign retention " "or an approved coordinated correction instead of rewriting one row." ), executable=False, ) ) if ( record.resource_type == "campaign_share" and "target_id" in match_fields and record.data.get("revoked_at") is None and subject_user_id is not None ): actions.append( _action( f"campaigns:revoke:campaign_share:{record.resource_id}", "revoke", record, "Revoke active Campaign share", "The active subject-targeted workspace share can be revoked without altering delivery evidence.", executable=True, metadata={ "subject_user_id": subject_user_id, "tenant_id": tenant_id, }, ) ) if ( record.resource_type == "recipient_import_mapping_profile" and "owner_user_id" in match_fields and subject_user_id is not None ): actions.append( _action( f"campaigns:delete:recipient_import_mapping_profile:{record.resource_id}", "delete", record, "Delete personal recipient import mapping", "The user-owned import preference can be removed without changing Campaign or delivery evidence.", executable=True, irreversible=True, metadata={ "subject_user_id": subject_user_id, "tenant_id": tenant_id, }, ) ) ids = [action.action_id for action in actions] if len(ids) != len(set(ids)): raise ValueError("Campaign DSAR produced duplicate action ids.") return tuple(actions) def execute_erasure( self, session: object, *, tenant_id: str, subject: DsarSubjectRef, actions: Sequence[DsarErasureActionRef], request_id: str, ) -> Sequence[DsarExecutionResultRef]: db = _session(session) subject_user_id = _subject_user_id(subject) results: list[DsarExecutionResultRef] = [] for action in actions: if ( subject_user_id is None or action.provider_id != self.provider_id or action.module_id != self.module_id or action.metadata.get("subject_user_id") != subject_user_id or action.metadata.get("tenant_id") != tenant_id ): results.append( _blocked(action, "The Campaign DSAR action is stale or invalid.") ) continue if action.action_id.startswith( "campaigns:delete:recipient_import_mapping_profile:" ): results.append( _delete_mapping_profile( db, tenant_id=tenant_id, subject_user_id=subject_user_id, action=action, request_id=request_id, ) ) elif action.action_id.startswith("campaigns:revoke:campaign_share:"): results.append( _revoke_share( db, tenant_id=tenant_id, subject_user_id=subject_user_id, action=action, request_id=request_id, ) ) else: results.append( _blocked(action, "Campaign does not execute this action kind.") ) db.flush() return tuple(results) def _append_delivery_evidence( self, db: Session, *, append: object, tenant_id: str, job_ids: set[str], immutable_jobs: Mapping[str, bool], ) -> None: if not job_ids: return issues = _bounded_rows( db.query(CampaignIssue) .filter( CampaignIssue.tenant_id == tenant_id, CampaignIssue.job_id.in_(job_ids), ) .order_by(CampaignIssue.id) ) for issue in issues: append( # type: ignore[operator] _record( "campaign_delivery_issue", issue.id, "campaign_delivery_evidence", issue.code, { "campaign_id": issue.campaign_id, "campaign_version_id": issue.campaign_version_id, "job_id": issue.job_id, "severity": issue.severity, "code": issue.code, "behavior": issue.behavior, }, observed_at=issue.updated_at, immutable=bool(immutable_jobs.get(str(issue.job_id))), retention_reason=( "Issues attached to an evidence-bearing delivery are retained with that delivery." if immutable_jobs.get(str(issue.job_id)) else None ), ) ) evidence_specs = ( ( SendAttempt, "campaign_send_attempt", lambda row: { "job_id": row.job_id, "attempt_number": row.attempt_number, "status": row.status, "smtp_status_code": row.smtp_status_code, "error_type": row.error_type, "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), }, ), ( ImapAppendAttempt, "campaign_imap_attempt", lambda row: { "job_id": row.job_id, "attempt_number": row.attempt_number, "status": row.status, }, ), ( PostboxDeliveryAttempt, "campaign_postbox_attempt", lambda row: { "job_id": row.job_id, "attempt_number": row.attempt_number, "status": row.status, "provider_delivery_id": row.provider_delivery_id, "provider_message_id": row.provider_message_id, "postbox_id": row.postbox_id, "address": row.address, "vacant": row.vacant, "duplicate": row.duplicate, "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), }, ), ( PrintOutputAttempt, "campaign_print_attempt", lambda row: { "job_id": row.job_id, "attempt_number": row.attempt_number, "status": row.status, "render_id": row.render_id, "artifact_sha256": row.artifact_sha256, "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), }, ), ) for model, resource_type, encoder in evidence_specs: for row in _bounded_rows( db.query(model).filter(model.job_id.in_(job_ids)).order_by(model.id) ): append( # type: ignore[operator] _record( resource_type, row.id, "campaign_delivery_evidence", resource_type.replace("_", " "), encoder(row), observed_at=row.updated_at, immutable=True, retention_reason="Provider-attempt outcomes are immutable delivery evidence.", ) ) actions = _bounded_rows( db.query(CampaignMessageAction) .filter( CampaignMessageAction.tenant_id == tenant_id, CampaignMessageAction.job_id.in_(job_ids), ) .order_by(CampaignMessageAction.id) ) action_ids = {row.id for row in actions} for row in actions: append( # type: ignore[operator] _record( "campaign_message_action", row.id, "campaign_delivery_evidence", f"Message action {row.kind}", { "job_id": row.job_id, "kind": row.kind, "status": row.status, "message_sha256": row.message_sha256, "recipient_count": row.recipient_count, "accepted_count": row.accepted_count, "refused_count": row.refused_count, "completed_at": _iso(row.completed_at), }, observed_at=row.updated_at, immutable=True, retention_reason="Single-message actions are immutable delivery and correction evidence.", ) ) if action_ids: for attempt in _bounded_rows( db.query(CampaignMessageActionAttempt) .filter(CampaignMessageActionAttempt.action_id.in_(action_ids)) .order_by(CampaignMessageActionAttempt.id) ): append( # type: ignore[operator] _record( "campaign_message_action_attempt", attempt.id, "campaign_delivery_evidence", "Message action attempt", { "action_id": attempt.action_id, "attempt_number": attempt.attempt_number, "status": attempt.status, "outcome_code": attempt.outcome_code, "started_at": _iso(attempt.started_at), "completed_at": _iso(attempt.completed_at), }, observed_at=attempt.updated_at, immutable=True, retention_reason="Action attempts are immutable correction evidence.", ) ) def _append_report_projections( self, *, append: object, jobs: Sequence[CampaignJob], ) -> None: by_campaign: dict[str, list[CampaignJob]] = {} for job in jobs: by_campaign.setdefault(job.campaign_id, []).append(job) for campaign_id, rows in sorted(by_campaign.items()): append( # type: ignore[operator] _record( "campaign_report_projection", campaign_id, "campaign_report", "Recipient-specific Campaign report projection", { "campaign_id": campaign_id, "matched_job_count": len(rows), "send_status_counts": dict( sorted(Counter(row.send_status for row in rows).items()) ), "postbox_status_counts": dict( sorted(Counter(row.postbox_status for row in rows).items()) ), "print_status_counts": dict( sorted(Counter(row.print_status for row in rows).items()) ), }, immutable=True, retention_reason=( "Campaign reports are derived projections over retained recipient and delivery evidence." ), source_path=f"/campaigns/{campaign_id}/report", ) ) def _append_member_resources( self, db: Session, *, append: object, tenant_id: str, subject_user_id: str | None, subject_account_id: str | None, campaign_ids: set[str], version_ids: set[str], ) -> None: assignment_filters = [] if subject_user_id is not None: assignment_filters.append( CampaignWorkAssignment.assigned_by_user_id == subject_user_id ) if subject_account_id is not None: assignment_filters.append( and_( CampaignWorkAssignment.assignee_type == "account", CampaignWorkAssignment.assignee_id == subject_account_id, ) ) assignment_ids: set[str] = set() if assignment_filters: assignment_rows = _bounded_rows( db.query(CampaignWorkAssignment) .filter( CampaignWorkAssignment.tenant_id == tenant_id, or_(*assignment_filters), ) .order_by(CampaignWorkAssignment.id) ) assignment_ids = {item.id for item in assignment_rows} for assignment in assignment_rows: match_fields = [] if assignment.assigned_by_user_id == subject_user_id: match_fields.append("assigned_by_user_id") if ( assignment.assignee_type == "account" and assignment.assignee_id == subject_account_id ): match_fields.append("assignee_id") append( # type: ignore[operator] _record( "campaign_work_assignment", assignment.id, "campaign_work_accountability", "Campaign work assignment", { "match_fields": match_fields, "campaign_id": assignment.campaign_id, "campaign_version_id": assignment.campaign_version_id, "purpose": assignment.purpose, "status": assignment.status, "due_at": _iso(assignment.due_at), "assignee_type": assignment.assignee_type, "assignee_id": assignment.assignee_id, "assignee_label_snapshot": assignment.assignee_label_snapshot, "assignee_resolution_state": assignment.assignee_resolution_state, "reference_kind": assignment.reference_kind, "reference_id": assignment.reference_id, "completed_at": _iso(assignment.completed_at), "cancelled_at": _iso(assignment.cancelled_at), }, observed_at=assignment.updated_at, source_path=f"/campaigns/{assignment.campaign_id}/work", ) ) event_filters = [] if subject_user_id is not None: event_filters.append( CampaignWorkAssignmentEvent.actor_user_id == subject_user_id ) if assignment_ids: event_filters.append( CampaignWorkAssignmentEvent.assignment_id.in_(assignment_ids) ) if subject_account_id is not None: event_filters.extend( ( and_( CampaignWorkAssignmentEvent.assignee_type_snapshot == "account", CampaignWorkAssignmentEvent.assignee_id_snapshot == subject_account_id, ), cast(CampaignWorkAssignmentEvent.details, Text).contains( subject_account_id ), ) ) if event_filters: event_rows = _bounded_rows( db.query(CampaignWorkAssignmentEvent) .filter( CampaignWorkAssignmentEvent.tenant_id == tenant_id, or_(*event_filters), ) .order_by(CampaignWorkAssignmentEvent.id) ) for event in event_rows: append( # type: ignore[operator] _record( "campaign_work_assignment_event", event.id, "campaign_work_accountability_evidence", "Campaign work assignment event", { "campaign_id": event.campaign_id, "assignment_id": event.assignment_id, "event_kind": event.event_kind, "status": event.status_snapshot, "assignee_type": event.assignee_type_snapshot, "assignee_id": event.assignee_id_snapshot, "assignee_label_snapshot": event.assignee_label_snapshot, "resolution_state": event.resolution_state_snapshot, }, observed_at=event.created_at, immutable=True, retention_reason="Assignment lifecycle events retain accountable institutional work history.", source_path=f"/campaigns/{event.campaign_id}/work", ) ) if subject_user_id is not None: collaboration_rows = _bounded_rows( db.query(CampaignCollaborationEntry) .filter( CampaignCollaborationEntry.tenant_id == tenant_id, or_( CampaignCollaborationEntry.actor_user_id == subject_user_id, CampaignCollaborationEntry.withdrawn_by_user_id == subject_user_id, CampaignCollaborationEntry.redacted_by_user_id == subject_user_id, cast(CampaignCollaborationEntry.mention_user_ids, Text).contains( f'"{subject_user_id}"' ), ), ) .order_by(CampaignCollaborationEntry.id) ) for entry in collaboration_rows: match_fields = _matching_fields( entry, subject_user_id, ( "actor_user_id", "withdrawn_by_user_id", "redacted_by_user_id", ), ) if subject_user_id in (entry.mention_user_ids or []): match_fields.append("mention_user_ids") authored = entry.actor_user_id == subject_user_id append( # type: ignore[operator] _record( "campaign_collaboration_entry", entry.id, "campaign_collaboration_evidence", "Campaign collaboration entry", { "match_fields": match_fields, "campaign_id": entry.campaign_id, "campaign_version_id": entry.campaign_version_id, "visibility": entry.visibility, "reference_kind": entry.reference_kind, "reference_id": entry.reference_id, "posted_text": entry.content if authored else None, "content_disclosed": authored and entry.content is not None, "content_sha256": entry.content_sha256, "withdrawn_at": _iso(entry.withdrawn_at), "redacted_at": _iso(entry.redacted_at), }, observed_at=entry.updated_at, immutable=True, retention_reason=( "Append-only discussion identity, reference, hash, tombstone, and moderation evidence are retained with Campaign access and Audit evidence." ), source_path=f"/campaigns/{entry.campaign_id}/activity", ) ) shares = _bounded_rows( db.query(CampaignShare) .filter( CampaignShare.tenant_id == tenant_id, or_( (CampaignShare.target_type == "user") & (CampaignShare.target_id == subject_user_id), CampaignShare.created_by_user_id == subject_user_id, ), ) .order_by(CampaignShare.id) ) for share in shares: match_fields = _matching_fields( share, subject_user_id, ("created_by_user_id",) ) if share.target_type == "user" and share.target_id == subject_user_id: match_fields.insert(0, "target_id") append( # type: ignore[operator] _record( "campaign_share", share.id, "campaign_access_evidence", f"Campaign share {share.id}", { "match_fields": match_fields, "campaign_id": share.campaign_id, "permission": share.permission, "revoked_at": _iso(share.revoked_at), }, observed_at=share.updated_at, immutable=True, retention_reason="Campaign sharing history is institutional access evidence.", source_path=f"/campaigns/{share.campaign_id}", ) ) for profile in _bounded_rows( db.query(RecipientImportMappingProfile) .filter( RecipientImportMappingProfile.tenant_id == tenant_id, RecipientImportMappingProfile.owner_user_id == subject_user_id, ) .order_by(RecipientImportMappingProfile.id) ): append( # type: ignore[operator] _record( "recipient_import_mapping_profile", profile.id, "personal_campaign_preference", profile.name, { "match_fields": ["owner_user_id"], "name": profile.name, "column_count": profile.column_count, "headers": _safe_value(profile.headers), "mappings": _safe_value(profile.mappings), "delimiter": profile.delimiter, "header_rows": profile.header_rows, }, observed_at=profile.updated_at, ) ) schedule_conditions = [] if subject_user_id is not None: schedule_conditions.append( CampaignSchedule.created_by_user_id == subject_user_id ) if version_ids: schedule_conditions.append( CampaignSchedule.source_version_id.in_(version_ids) ) if schedule_conditions: for schedule in _bounded_rows( db.query(CampaignSchedule) .filter( CampaignSchedule.tenant_id == tenant_id, or_(*schedule_conditions), ) .order_by(CampaignSchedule.id) ): append( # type: ignore[operator] _record( "campaign_schedule", schedule.id, "campaign_schedule_evidence", schedule.name, { "match_fields": _matching_fields( schedule, subject_user_id, ("created_by_user_id",) ) + ( ["source_version.recipient"] if schedule.source_version_id in version_ids else [] ), "campaign_id": schedule.campaign_id, "source_version_id": schedule.source_version_id, "delivery_mode": schedule.delivery_mode, "recurrence_kind": schedule.recurrence_kind, "active": schedule.active, "starts_at": _iso(schedule.starts_at), "next_fire_at": _iso(schedule.next_fire_at), }, observed_at=schedule.updated_at, immutable=True, retention_reason="Schedule configuration and occurrence intent are execution evidence.", source_path=f"/campaigns/{schedule.campaign_id}", ) ) attachment_conditions = [] if subject_user_id is not None: attachment_conditions.append( AttachmentInstance.owner_user_id == subject_user_id, ) if campaign_ids: attachment_conditions.append( AttachmentInstance.campaign_id.in_(campaign_ids) ) if not attachment_conditions: return attachment_query = db.query(AttachmentInstance).filter( AttachmentInstance.tenant_id == tenant_id, or_(*attachment_conditions), ) for attachment in _bounded_rows( attachment_query.order_by(AttachmentInstance.id) ): append( # type: ignore[operator] _record( "campaign_attachment", attachment.id, "campaign_artifact", attachment.filename, { "match_fields": _matching_fields( attachment, subject_user_id, ("owner_user_id",) ) + ( ["campaign.recipient"] if attachment.campaign_id in campaign_ids else [] ), "campaign_id": attachment.campaign_id, "logical_name": attachment.logical_name, "filename": attachment.filename, "tags": _safe_value(attachment.tags), }, observed_at=attachment.updated_at, immutable=attachment.campaign_id in campaign_ids, retention_reason=( "Attachments associated with recipient delivery follow Campaign artifact retention." if attachment.campaign_id in campaign_ids else None ), ) ) def _matching_jobs( session: Session, *, tenant_id: str, email: str | None, job_id: str | None, entry_id: str | None, ) -> list[CampaignJob]: candidate_sets: list[set[str]] = [] if email: candidate_sets.append( { row[0] for row in _bounded_rows( session.query(CampaignJob.id) .filter( CampaignJob.tenant_id == tenant_id, func.lower(CampaignJob.recipient_email) == email, ) .order_by(CampaignJob.id) ) } ) if job_id: candidate_sets.append( { row[0] for row in session.query(CampaignJob.id).filter( CampaignJob.tenant_id == tenant_id, CampaignJob.id == job_id, ) } ) if entry_id: candidate_sets.append( { row[0] for row in _bounded_rows( session.query(CampaignJob.id) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.entry_id == entry_id, ) .order_by(CampaignJob.id) ) } ) if not candidate_sets: return [] ids = set.intersection(*candidate_sets) if not ids: return [] return _bounded_rows( session.query(CampaignJob) .filter(CampaignJob.tenant_id == tenant_id, CampaignJob.id.in_(ids)) .order_by(CampaignJob.id) ) def _matching_versions( session: Session, *, tenant_id: str, subject_user_id: str | None, email: str | None, job_version_ids: set[str], version_id: str | None, ) -> list[CampaignVersion]: conditions = [] if job_version_ids: conditions.append(CampaignVersion.id.in_(job_version_ids)) if version_id: conditions.append(CampaignVersion.id == version_id) if subject_user_id: conditions.extend( ( CampaignVersion.locked_by_user_id == subject_user_id, CampaignVersion.user_locked_by_user_id == subject_user_id, CampaignVersion.archived_by_user_id == subject_user_id, ) ) if email: pattern = f"%{_escape_like(email)}%" conditions.append( func.lower(cast(CampaignVersion.raw_json, Text)).like(pattern, escape="\\") ) if not conditions: return [] candidates = _bounded_rows( session.query(CampaignVersion) .join(Campaign, Campaign.id == CampaignVersion.campaign_id) .filter(Campaign.tenant_id == tenant_id, or_(*conditions)) .order_by(CampaignVersion.id) ) return [ row for row in candidates if row.id in job_version_ids or row.id == version_id or ( subject_user_id is not None and any( getattr(row, field) == subject_user_id for field in ( "locked_by_user_id", "user_locked_by_user_id", "archived_by_user_id", ) ) ) or bool(_matching_entry_fragments(row.raw_json, email)) ] def _matching_campaigns( session: Session, *, tenant_id: str, subject_user_id: str | None, campaign_ids: set[str], ) -> list[Campaign]: conditions = [] if campaign_ids: conditions.append(Campaign.id.in_(campaign_ids)) if subject_user_id: conditions.extend( ( Campaign.created_by_user_id == subject_user_id, Campaign.owner_user_id == subject_user_id, ) ) if not conditions: return [] return _bounded_rows( session.query(Campaign) .filter(Campaign.tenant_id == tenant_id, or_(*conditions)) .order_by(Campaign.id) ) def _matching_entry_fragments(value: object, email: str | None) -> list[object]: if not email or not isinstance(value, Mapping): return [] entries_value = value.get("entries") if isinstance(entries_value, Mapping): entries_value = entries_value.get("inline") if not isinstance(entries_value, list): return [] fragments: list[object] = [] for index, entry in enumerate(entries_value): paths = _matching_email_paths(entry, email=email) if not paths: continue fragments.append( { "entry_index": index, "entry_id": entry.get("id") if isinstance(entry, Mapping) else None, "matched_fields": paths, "data": _safe_value(entry, subject_email=email), } ) return fragments def _matching_recipient_fragment(value: object, email: str | None) -> object: if not email or not isinstance(value, Mapping): return {} return _safe_value(value, subject_email=email) def _matching_email_paths( value: object, *, email: str, path: str = "entry", depth: int = 0, ) -> list[str]: if depth >= 8: return [] matches: list[str] = [] if isinstance(value, Mapping): for key, item in list(value.items())[:128]: child = f"{path}.{key}" if str(key).casefold() == "email" and _normalized_email(item) == email: matches.append(child) else: matches.extend( _matching_email_paths( item, email=email, path=child, depth=depth + 1, ) ) elif isinstance(value, list): for index, item in enumerate(value[:256]): matches.extend( _matching_email_paths( item, email=email, path=f"{path}[{index}]", depth=depth + 1, ) ) return matches[:256] def _safe_value( value: object, *, subject_email: str | None = None, depth: int = 0, ) -> object: if depth >= 6: return "[depth-limited]" if isinstance(value, Mapping): item_email = _normalized_email(value.get("email")) if subject_email and item_email and item_email != subject_email: return None result: dict[str, object] = {} for key, item in list(value.items())[:128]: normalized_key = str(key).casefold() if ( any(part in normalized_key for part in _SECRET_PARTS) or normalized_key in _CONTENT_KEYS or normalized_key.endswith("_path") or normalized_key.endswith("_storage_key") ): continue sanitized = _safe_value( item, subject_email=subject_email, depth=depth + 1, ) if sanitized is not None: result[str(key)] = sanitized return result if isinstance(value, list): result = [] for item in value[:256]: sanitized = _safe_value( item, subject_email=subject_email, depth=depth + 1, ) if sanitized is not None: result.append(sanitized) return result if isinstance(value, str): if subject_email and "@" in value and _normalized_email(value) != subject_email: return None return value[:2_000] if value is None or isinstance(value, (bool, int, float)): return value return str(value)[:2_000] def _version_is_evidence(version: CampaignVersion, matched_jobs: int) -> bool: return bool( matched_jobs or version.workflow_state != "editing" or version.published_at or version.locked_at or version.archived_at or version.execution_snapshot_hash ) def _job_is_evidence(job: CampaignJob) -> bool: return bool( job.build_status != "pending" or job.queue_status != "draft" or job.send_status != "not_queued" or job.postbox_status != "not_requested" or job.print_status != "not_requested" or job.imap_status != "not_requested" or job.eml_sha256 or job.eml_storage_key or job.eml_local_path ) def _delete_mapping_profile( session: Session, *, tenant_id: str, subject_user_id: str, action: DsarErasureActionRef, request_id: str, ) -> DsarExecutionResultRef: row = ( session.query(RecipientImportMappingProfile) .filter( RecipientImportMappingProfile.id == action.resource_id, RecipientImportMappingProfile.tenant_id == tenant_id, ) .with_for_update() .one_or_none() ) if row is None: return _result( action, "unchanged", "The recipient import mapping was already absent.", {"request_id": request_id}, ) if row.owner_user_id != subject_user_id: return _blocked(action, "The mapping owner changed after planning.") session.delete(row) return _result( action, "executed", "The personal recipient import mapping was deleted.", {"request_id": request_id}, ) def _revoke_share( session: Session, *, tenant_id: str, subject_user_id: str, action: DsarErasureActionRef, request_id: str, ) -> DsarExecutionResultRef: row = ( session.query(CampaignShare) .filter(CampaignShare.id == action.resource_id) .with_for_update() .one_or_none() ) if ( row is None or row.tenant_id != tenant_id or row.target_type != "user" or row.target_id != subject_user_id ): return _blocked( action, "The subject-targeted Campaign share is no longer available." ) if row.revoked_at is not None: return _result( action, "unchanged", "The Campaign share was already revoked.", {"request_id": request_id, "revoked_at": _iso(row.revoked_at)}, ) row.revoked_at = datetime.now(timezone.utc) return _result( action, "executed", "The subject-targeted Campaign share was revoked.", {"request_id": request_id, "revoked_at": _iso(row.revoked_at)}, ) def _campaign_references(subject: DsarSubjectRef) -> dict[str, str]: aliases = { "campaign.campaign": "campaign", "campaign.version": "version", "campaign.job": "job", "campaign.entry": "entry", } result: dict[str, str] = {} for key, target in aliases.items(): value = str(subject.external_references.get(key) or "").strip() if value: result[target] = value return result def _subject_user_id(subject: DsarSubjectRef) -> str | None: candidates: list[str] = [] if subject.membership_id: candidates.append(subject.membership_id) for key in ( "campaign.user", "campaign.membership", "access.membership", "membership_id", ): value = str(subject.external_references.get(key) or "").strip() if value: candidates.append(value) normalized = {value.strip() for value in candidates if value.strip()} return normalized.pop() if len(normalized) == 1 else None def _subject_email(subject: DsarSubjectRef) -> str | None: candidates = [] if subject.email: candidates.append(subject.email) for key in ("campaign.email", "campaign.recipient_email"): value = str(subject.external_references.get(key) or "").strip() if value: candidates.append(value) normalized = { email for value in candidates if (email := _normalized_email(value)) is not None } return normalized.pop() if len(normalized) == 1 else None def _normalized_email(value: object) -> str | None: if not isinstance(value, str): return None normalized = value.strip().casefold() return normalized or None def _matching_fields( row: object, subject_user_id: str | None, fields: Sequence[str], ) -> list[str]: if subject_user_id is None: return [] return [field for field in fields if getattr(row, field) == subject_user_id] def _record( resource_type: str, resource_id: str, category: str, title: str, data: Mapping[str, object], *, observed_at: datetime | None = None, immutable: bool = False, retention_reason: str | None = None, source_path: str | None = None, ) -> DsarRecordRef: return DsarRecordRef( provider_id="campaigns", module_id="campaigns", resource_type=resource_type, resource_id=resource_id, category=category, title=title, data=data, observed_at=observed_at, immutable_evidence=immutable, retention_reason=retention_reason, source_path=source_path, ) def _action( action_id: str, kind: str, record: DsarRecordRef, title: str, rationale: str, *, executable: bool, irreversible: bool = False, metadata: Mapping[str, object] | None = None, ) -> DsarErasureActionRef: return DsarErasureActionRef( action_id=action_id, provider_id="campaigns", module_id="campaigns", kind=kind, # type: ignore[arg-type] resource_type=record.resource_type, resource_id=record.resource_id, title=title, rationale=rationale, executable=executable, irreversible=irreversible, metadata=metadata or {}, ) def _result( action: DsarErasureActionRef, status: str, summary: str, evidence: Mapping[str, object] | None = None, ) -> DsarExecutionResultRef: return DsarExecutionResultRef( action_id=action.action_id, status=status, # type: ignore[arg-type] summary=summary, evidence=evidence or {}, ) def _blocked(action: DsarErasureActionRef, summary: str) -> DsarExecutionResultRef: return _result(action, "blocked", summary) def _session(value: object) -> Session: if not isinstance(value, Session): raise TypeError("Campaign DSAR provider requires a SQLAlchemy session.") return value def _bounded_rows(query: object) -> list[object]: rows = query.limit(_MAX_RECORDS + 1).all() # type: ignore[attr-defined] if len(rows) > _MAX_RECORDS: raise ValueError( "Campaign DSAR match limit exceeded; narrow the subject selectors." ) return rows def _escape_like(value: str) -> str: return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _iso(value: datetime | None) -> str | None: if value is None: return None if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.isoformat() __all__ = ["CAMPAIGN_DSAR_CAPABILITY", "CampaignDsarProvider"]