multi-seal-mail/server/app/storage/campaign_usage.py

from __future__ import annotations

from pathlib import PurePosixPath

from sqlalchemy.orm import Session

from app.db.models import CampaignAttachmentUse, CampaignJob, FileAsset, FileBlob, FileVersion
from app.storage.common import utcnow
from app.storage.files import current_version_and_blob, list_assets_for_user


def _candidate_match_keys(raw_match: str) -> set[str]:
    cleaned = raw_match.replace("\\", "/").strip().strip("/")
    result = {cleaned}
    if cleaned:
        result.add(PurePosixPath(cleaned).name)
    return {item for item in result if item}


def _add_use(
    session: Session,
    job: CampaignJob,
    *,
    asset: FileAsset,
    version: FileVersion,
    blob: FileBlob,
    filename_used: str,
    stage: str,
) -> None:
    exists = (
        session.query(CampaignAttachmentUse)
        .filter(
            CampaignAttachmentUse.campaign_job_id == job.id,
            CampaignAttachmentUse.file_version_id == version.id,
            CampaignAttachmentUse.filename_used == filename_used,
            CampaignAttachmentUse.use_stage == stage,
        )
        .one_or_none()
    )
    if exists:
        return
    session.add(
        CampaignAttachmentUse(
            tenant_id=job.tenant_id,
            campaign_id=job.campaign_id,
            campaign_version_id=job.campaign_version_id,
            campaign_job_id=job.id,
            entry_index=job.entry_index,
            entry_id=job.entry_id,
            file_asset_id=asset.id,
            file_version_id=version.id,
            file_blob_id=blob.id,
            filename_used=filename_used,
            checksum_sha256=blob.checksum_sha256,
            size_bytes=blob.size_bytes,
            content_type=blob.content_type,
            use_stage=stage,
        )
    )


def record_campaign_attachment_uses_for_job(session: Session, job: CampaignJob, *, stage: str = "built") -> None:
    """Record immutable managed file versions used by a built/sent job.

    New builds carry exact managed asset/version/blob IDs. Filename matching is
    retained only as a compatibility fallback for jobs created before managed
    attachment materialization was introduced.
    """

    attachments = job.resolved_attachments or []
    if not isinstance(attachments, list):
        return

    assets = list_assets_for_user(
        session,
        tenant_id=job.tenant_id,
        user_id="",
        campaign_id=job.campaign_id,
        is_admin=True,
    )
    assets_by_id = {asset.id: asset for asset in assets}

    for attachment in attachments:
        if not isinstance(attachment, dict):
            continue
        managed_matches = attachment.get("managed_matches")
        if isinstance(managed_matches, list):
            for item in managed_matches:
                if not isinstance(item, dict):
                    continue
                asset = assets_by_id.get(str(item.get("asset_id") or ""))
                version = session.get(FileVersion, str(item.get("version_id") or ""))
                blob = session.get(FileBlob, str(item.get("blob_id") or ""))
                if not asset or not version or not blob:
                    continue
                if version.file_asset_id != asset.id or version.blob_id != blob.id:
                    continue
                _add_use(
                    session,
                    job,
                    asset=asset,
                    version=version,
                    blob=blob,
                    filename_used=str(item.get("filename") or asset.filename),
                    stage=stage,
                )

    # Compatibility fallback for older job snapshots without managed_matches.
    by_key: dict[str, FileAsset] = {}
    for asset in assets:
        by_key[asset.display_path.strip("/")] = asset
        by_key.setdefault(asset.filename, asset)
    for attachment in attachments:
        if not isinstance(attachment, dict) or attachment.get("managed_matches"):
            continue
        matches = attachment.get("matches") if isinstance(attachment.get("matches"), list) else []
        for raw in matches:
            if not isinstance(raw, str):
                continue
            asset = next((by_key[key] for key in _candidate_match_keys(raw) if key in by_key), None)
            if not asset:
                continue
            version, blob = current_version_and_blob(session, asset)
            _add_use(
                session,
                job,
                asset=asset,
                version=version,
                blob=blob,
                filename_used=asset.filename,
                stage=stage,
            )


def mark_job_attachment_uses_sent(session: Session, job: CampaignJob) -> None:
    record_campaign_attachment_uses_for_job(session, job, stage="built")
    now = utcnow()
    uses = (
        session.query(CampaignAttachmentUse)
        .filter(
            CampaignAttachmentUse.tenant_id == job.tenant_id,
            CampaignAttachmentUse.campaign_job_id == job.id,
            CampaignAttachmentUse.use_stage == "built",
        )
        .all()
    )
    for use in uses:
        sent = (
            session.query(CampaignAttachmentUse)
            .filter(
                CampaignAttachmentUse.campaign_job_id == job.id,
                CampaignAttachmentUse.file_version_id == use.file_version_id,
                CampaignAttachmentUse.use_stage == "sent",
            )
            .one_or_none()
        )
        if sent:
            continue
        session.add(
            CampaignAttachmentUse(
                tenant_id=use.tenant_id,
                campaign_id=use.campaign_id,
                campaign_version_id=use.campaign_version_id,
                campaign_job_id=use.campaign_job_id,
                entry_index=use.entry_index,
                entry_id=use.entry_id,
                file_asset_id=use.file_asset_id,
                file_version_id=use.file_version_id,
                file_blob_id=use.file_blob_id,
                filename_used=use.filename_used,
                checksum_sha256=use.checksum_sha256,
                size_bytes=use.size_bytes,
                content_type=use.content_type,
                use_stage="sent",
                used_at=now,
            )
        )