"""Repository / Documents — service layer.

Functions are the only sanctioned mutators of FSM state. Views and APIs call
these; tests cover them. Each function wraps the change in `transaction.atomic`
so signal/audit/celery dispatch happen consistently after a successful commit.
"""

from __future__ import annotations

import hashlib
import logging
import os
import re
import secrets
import shutil
import uuid
from datetime import timedelta
from pathlib import Path
from typing import Any

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.files import File as DjangoFile
from django.db import models, transaction
from django.utils import timezone
from django.utils.text import get_valid_filename

from apps.core.notifications.models import Notification
from apps.core.notifications.services import send_notification
from apps.core.permissions.policies import grant_object_perm, revoke_object_perm
from apps.core.permissions.roles import UserRole
from apps.core.storage import UploadHandler
from apps.core.storage.registry import (
    DEFAULT_MAX_SIZE_MB,
    FILE_TYPE_REGISTRY,
    MODULE_SIZE_LIMITS,
)
from apps.repository.documents import signals as repo_signals
from apps.repository.documents.audit_helper import (
    action_create,
    action_delete,
    action_update,
    audit_repository,
)
from apps.repository.documents.models import (
    AccessRequest,
    Author,
    AuthorAlias,
    Collection,
    CollectionGroupAccess,
    CollectionSlugAlias,
    Document,
    DocumentAuthor,
    DocumentComment,
    DocumentReviewTask,
    DocumentShare,
    DocumentVersion,
    IntegrationOutbox,
    Keyword,
    QARecord,
    ResumableUpload,
)


class IncompatibleMergeError(ValidationError):
    """Raised when two records cannot be merged — e.g. visibility mismatch."""

logger = logging.getLogger(__name__)
User = get_user_model()


# ── helpers ────────────────────────────────────────────────────────────────


# Categories the repository accepts. Audio/video ("media") is intentionally
# excluded — the repository is a document archive, not a media library (#2,
# FRREP-DCI005). "archive" stays in for ZIP bundles of related artefacts.
REPOSITORY_UPLOAD_CATEGORIES = ["document", "data", "image", "archive"]


def _is_malware_scan_failure(exc: ValidationError) -> bool:
    """True when ``exc`` is the ClamAV-positive raised by ``scan_stream_optional``."""
    try:
        messages = exc.messages
    except AttributeError:  # pragma: no cover — defensive
        messages = [str(exc)]
    return any("virus scan" in str(m).lower() for m in messages)


def _report_malware_detection(file, exc: ValidationError, *, actor=None, request=None) -> None:
    """FRREP-DCI007 — a ClamAV positive must not fail silently. Audit it and
    alert Repository Administrators so the infected upload gets investigated,
    instead of only surfacing a rejection to the uploader."""
    filename = getattr(file, "name", "") or "(unnamed upload)"
    logger.warning("malware scan blocked upload %r: %s", filename, exc)
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=None,
        object_repr=filename[:200],
        changes={
            "malware_scan": "positive",
            "filename": filename,
            "detail": "; ".join(exc.messages) if hasattr(exc, "messages") else str(exc),
        },
        request=request,
    )
    admins = list(User.objects.filter(role=UserRole.REPO_MANAGER, is_active=True))
    if admins:
        from apps.core.notifications.services import bulk_notify

        bulk_notify(
            admins,
            f'Virus scan blocked an upload: "{filename}".',
            verb=Notification.Verb.REPO_UPLOAD_MALWARE_DETECTED,
        )


def _check_pdf_structure(file) -> None:
    """Structural sanity check for PDFs (FRREP-DCI008).

    A full ``pypdf`` object-graph parse (walking the xref table) is
    deliberately *not* used here: it rejects a large class of PDFs that are
    perfectly fine for archival purposes — e.g. files produced by tools that
    write a non-standard-but-linear object layout — and pypdf has no repair
    path when the xref table itself is missing. What actually indicates a
    truncated/corrupted upload (a network interruption mid-transfer, a
    non-PDF file that slipped past the MIME check, a zero-byte stub) is the
    absence of the ``%PDF-`` header and/or the trailing ``%%EOF`` marker that
    every complete PDF writer emits — so that's what we check.
    """
    file.seek(0)
    head = file.read(1024)
    if not head.lstrip().startswith(b"%PDF-"):
        raise ValidationError(
            "This file does not look like a valid PDF (missing the %PDF header)."
        )
    file.seek(0, os.SEEK_END)
    size = file.tell()
    file.seek(max(0, size - 2048))
    tail = file.read()
    if b"%%EOF" not in tail:
        raise ValidationError(
            "This PDF file appears to be truncated or corrupted — no end-of-file "
            "marker was found. Please re-export or re-upload the file."
        )


def _validate_file_structure(file) -> None:
    """FRREP-DCI008 — reject structurally corrupt PDF/DOCX files.

    ``extract_file_metadata`` treats a corrupt file as "no metadata found",
    which is the right behaviour for a best-effort *suggestion* extractor —
    but it means a corrupt file otherwise sails straight into the archive
    with no signal to the uploader. This runs a structural probe in
    validation mode: a PDF missing its header/trailer, or a DOCX zip
    container that fails to open, raises a clear ValidationError instead of
    silently accepting the file.
    """
    name = getattr(file, "name", "") or ""
    suffix = Path(name).suffix.lower()
    if suffix not in (".pdf", ".docx"):
        return
    pos = file.tell() if hasattr(file, "tell") else 0
    if hasattr(file, "seek"):
        file.seek(0)
    try:
        if suffix == ".pdf":
            _check_pdf_structure(file)
        elif suffix == ".docx":
            try:
                from docx import Document as DocxDocument
            except ImportError:  # pragma: no cover — optional dependency
                return
            try:
                DocxDocument(file)
            except Exception as exc:
                raise ValidationError(
                    f"This DOCX file appears to be corrupted and could not be read: {exc}"
                ) from exc
    finally:
        if hasattr(file, "seek"):
            file.seek(pos)


def _validate_repository_upload(file, *, actor=None, request=None):
    try:
        UploadHandler(
            module="repository",
            allowed_categories=REPOSITORY_UPLOAD_CATEGORIES,
        ).handle(file)
    except ValidationError as exc:
        if _is_malware_scan_failure(exc):
            _report_malware_detection(file, exc, actor=actor, request=request)
        raise
    _validate_file_structure(file)


def compute_file_hash(file) -> str:
    h = hashlib.sha256()
    pos = file.tell() if hasattr(file, "tell") else 0
    if hasattr(file, "seek"):
        file.seek(0)
    for chunk in iter(lambda: file.read(8192), b""):
        h.update(chunk)
    if hasattr(file, "seek"):
        file.seek(pos)
    return h.hexdigest()


def detect_duplicate(file_hash: str) -> Document | None:
    if not file_hash:
        return None
    return Document.objects.filter(file_hash=file_hash).first()


def _allocate_public_document_id() -> str:
    """Compact human-friendly ID like 'REPO-2026-XXXXXX'."""
    year = timezone.now().year
    suffix = secrets.token_hex(3).upper()
    candidate = f"REPO-{year}-{suffix}"
    while Document.objects.filter(public_document_id=candidate).exists():
        suffix = secrets.token_hex(3).upper()
        candidate = f"REPO-{year}-{suffix}"
    return candidate


# ── automatic metadata extraction (FRREP-DCI009) ─────────────────────────────


def _looks_like_title(line: str) -> bool:
    """Heuristic: is this line a plausible document title?"""
    line = (line or "").strip()
    if not (10 <= len(line) <= 250):
        return False
    low = line.lower()
    if low.startswith((
        "abstract", "doi", "http", "www.", "keywords", "vol.", "volume",
        "page ", "©", "copyright", "table of contents", "contents", "issn", "isbn",
    )):
        return False
    # Reject lines that are clearly boilerplate / locators rather than a title.
    if "http" in low or "www." in low or "available from" in low or "@" in line:
        return False
    # Reject lines that are mostly digits/punctuation (page numbers, dates, refs).
    letters = sum(ch.isalpha() for ch in line)
    return letters >= len(line) * 0.6


def _title_from_pdf_text(reader) -> str:
    """First title-like line on page 1 — used when the PDF has no embedded Title
    (the common case for real-world documents)."""
    try:
        text = reader.pages[0].extract_text() or ""
    except Exception:
        return ""
    for raw in text.splitlines():
        if _looks_like_title(raw):
            return raw.strip()[:400]
    return ""


def _title_from_docx(docx_obj) -> str:
    """A Title/Heading-styled paragraph, else the first title-like line."""
    try:
        paragraphs = list(docx_obj.paragraphs)
    except Exception:
        return ""
    for para in paragraphs:
        style = (getattr(para.style, "name", "") or "").lower()
        if ("title" in style or style.startswith("heading")) and (para.text or "").strip():
            return para.text.strip()[:400]
    for para in paragraphs:
        if _looks_like_title(para.text or ""):
            return para.text.strip()[:400]
    return ""


def _abstract_from_text(text: str) -> str:
    """Pull the abstract/summary paragraph out of a document's opening text.

    Looks for an ``Abstract`` / ``Summary`` heading and captures the prose that
    follows, stopping at the next section marker (Keywords, Introduction, …).
    Returns "" when nothing convincing is found — it's an editable suggestion.
    """
    if not text:
        return ""
    match = re.search(
        r"\b(?:abstract|executive\s+summary|summary)\b[\s:.–—\-]*\n?"
        r"(.+?)"
        r"(?:\n\s*\n|\bkey\s?words?\b|\bintroduction\b|\b1\.\s|\bj\.?e\.?l\.?\b|\bbackground\b|$)",
        text,
        re.IGNORECASE | re.DOTALL,
    )
    if match:
        body = re.sub(r"\s+", " ", match.group(1)).strip()
        if len(body) >= 120:  # a real abstract paragraph, not a stray heading
            return body[:3000]

    # Fallback: no explicit "Abstract" heading — recover the first run of
    # flowing prose lines as an editable starting point (better than a blank
    # required field). Works line-by-line so the title/author/affiliation lines
    # (which carry footnote markers or are digit-heavy) are skipped cleanly,
    # rather than being glued onto the abstract by whitespace-collapsing first.
    out: list[str] = []
    started = False
    for raw in text.splitlines():
        line = raw.strip()
        if not line:
            if started:
                break  # blank line ends the abstract paragraph
            continue
        if re.match(r"^(keywords?|introduction|references|acknowledg|table of|contents|\d+[.\s])", line, re.IGNORECASE):
            if started:
                break
            continue
        words = line.split()
        lower = sum(1 for w in words if w[:1].islower())
        has_marker = bool(re.search(r"[†‡§¶]", line))
        digit_ratio = sum(1 for w in words if any(c.isdigit() for c in w)) / max(1, len(words))
        looks_prose = (
            len(line) >= 45 and len(words) >= 7
            and lower >= len(words) * 0.5 and not has_marker and digit_ratio < 0.25
        )
        if not started:
            if looks_prose:
                started = True
                out.append(line)
            continue
        # In the abstract now: stop at an author/affiliation/heading-style line.
        if has_marker or digit_ratio > 0.4 or len(line) < 8:
            break
        out.append(line)
        if len(" ".join(out)) >= 3000:  # safety cap; abstracts rarely exceed this
            break
    body = re.sub(r"\s+", " ", " ".join(out)).strip()
    return body[:3000] if len(body) >= 120 else ""


def _pdf_first_text(reader, pages: int = 2) -> str:
    """Concatenated text of the first ``pages`` pages (where abstracts live)."""
    chunks: list[str] = []
    try:
        for page in reader.pages[:pages]:
            chunks.append(page.extract_text() or "")
    except Exception:
        pass
    return "\n".join(chunks)


def extract_file_metadata(path, *, filename: str = "") -> dict[str, Any]:
    """Best-effort extraction of embedded metadata from an uploaded file.

    Returns a dict with any of ``title``, ``author``, ``year`` it can read from
    the document's own properties (PDF info dict via pypdf, DOCX core-properties
    via python-docx), plus the always-available ``filename`` and ``size_bytes``.
    Missing values are simply omitted — callers treat the result as *editable
    suggestions*, never authoritative. Never raises: a corrupt or metadata-less
    file yields just the filesystem facts (FRREP-DCI009).
    """
    path = Path(path)
    out: dict[str, Any] = {}
    name = filename or path.name
    if name:
        out["filename"] = name
    try:
        out["size_bytes"] = path.stat().st_size
    except OSError:
        pass

    suffix = Path(name).suffix.lower()
    try:
        if suffix == ".pdf":
            from pypdf import PdfReader

            reader = PdfReader(str(path))
            info = reader.metadata or {}
            title = (getattr(info, "title", None) or "").strip()
            author = (getattr(info, "author", None) or "").strip()
            if not title:
                # Most real PDFs leave the embedded Title blank — recover it from
                # the first page's text so the suggestion is actually useful.
                title = _title_from_pdf_text(reader)
            if title:
                out["title"] = title[:400]
            if author:
                out["author"] = author[:255]
            abstract = _abstract_from_text(_pdf_first_text(reader))
            if abstract:
                out["abstract"] = abstract
            created = getattr(info, "creation_date", None)
            if created is not None and getattr(created, "year", None):
                out["year"] = created.year
        elif suffix == ".docx":
            from docx import Document as DocxDocument

            docx_obj = DocxDocument(str(path))
            props = docx_obj.core_properties
            title = (props.title or "").strip()
            author = (props.author or "").strip()
            if not title:
                title = _title_from_docx(docx_obj)
            if title:
                out["title"] = title[:400]
            if author:
                out["author"] = author[:255]
            docx_text = "\n".join(p.text or "" for p in docx_obj.paragraphs[:80])
            abstract = _abstract_from_text(docx_text)
            if abstract:
                out["abstract"] = abstract
            if props.created is not None and getattr(props.created, "year", None):
                out["year"] = props.created.year
    except Exception:  # pragma: no cover — extraction is best-effort
        logger.warning("metadata extraction failed for %s", name, exc_info=True)
    return out


# ── author registry ─────────────────────────────────────────────────────────


@transaction.atomic
def register_author(
    *,
    first_name: str = "",
    last_name: str = "",
    name: str | None = None,
    orcid: str | None = None,
    email: str = "",
    institution=None,
    affiliation_text: str = "",
) -> Author:
    if name and not (first_name or last_name):
        parts = name.strip().split(" ", 1)
        first_name = parts[0]
        last_name = parts[1] if len(parts) > 1 else ""

    first_name = (first_name or "").strip()
    last_name = (last_name or "").strip()
    email = (email or "").strip()
    orcid = (orcid or "").strip() or None

    # FRREP-MCL002 — `verified_at` means "confirmed against the ORCID public
    # API", not merely "an ORCID string was supplied". Only the branch where
    # `lookup_orcid` actually returns data sets this flag.
    orcid_verified = False
    if orcid:
        existing = Author.objects.filter(orcid=orcid).first()
        if existing:
            return existing
        from apps.repository.documents.ingestion import lookup_orcid

        lookup = lookup_orcid(orcid)
        if lookup:
            orcid_verified = True
            first_name = first_name or lookup.get("first_name", "")
            last_name = last_name or lookup.get("last_name", "")
            email = email or lookup.get("email", "")

    # ORCID is the strong identity. Without one we dedupe defensively:
    #   1. by case-insensitive email when present (a person typing the same
    #      email twice is almost certainly the same person);
    #   2. otherwise by case-insensitive (first_name, last_name) among ORCID-less
    #      rows — same-name collisions across institutions remain separate
    #      records and can be merged later via FRREP-MCL004;
    #   3. otherwise by `AuthorAlias.full_name` so historical names resolve to
    #      the canonical author (FRREP-MCL004 name-change linking).
    # We never look up `orcid=None` directly: that would return the first
    # ORCID-less row in the database regardless of name.
    author = None
    if orcid:
        author = Author.objects.filter(orcid=orcid).first()
    elif email:
        author = Author.objects.filter(orcid__isnull=True, email__iexact=email).first()
    elif first_name or last_name:
        author = Author.objects.filter(
            orcid__isnull=True,
            first_name__iexact=first_name,
            last_name__iexact=last_name,
        ).first()
        if author is None:
            full = f"{first_name} {last_name}".strip()
            if full:
                alias = (
                    AuthorAlias.objects.filter(full_name__iexact=full)
                    .select_related("author")
                    .first()
                )
                if alias and not alias.author.is_archived:
                    author = alias.author

    if author is None:
        author = Author.objects.create(
            first_name=first_name,
            last_name=last_name,
            email=email,
            orcid=orcid,
            institution=institution,
            affiliation_text=affiliation_text,
        )
        created = True
    else:
        created = False

    if not created:
        # Backfill any missing fields without overwriting populated values.
        dirty = []
        for field, value in (
            ("first_name", first_name),
            ("last_name", last_name),
            ("email", email),
            ("affiliation_text", affiliation_text),
        ):
            if value and not getattr(author, field):
                setattr(author, field, value)
                dirty.append(field)
        if institution and not author.institution_id:
            author.institution = institution
            dirty.append("institution")
        if dirty:
            author.save(update_fields=dirty + ["updated_at"])

    if orcid_verified and not author.verified_at:
        author.verified_at = timezone.now()
        author.save(update_fields=["verified_at", "updated_at"])
    return author


# ── ingestion ───────────────────────────────────────────────────────────────


class DuplicateDocumentError(ValidationError):
    """Raised when a file with the same hash already exists.

    The duplicate `Document` is exposed on `.duplicate_of` so the caller can
    offer the user the three options described in PRD REPO-SP1: cancel,
    upload as new version, or continue as a separate document.
    """

    def __init__(self, duplicate: Document):
        super().__init__("A document with the same file content already exists.")
        self.duplicate_of = duplicate


def ingest_document(
    *,
    file,
    title: str,
    collection: Collection,
    document_type: str,
    abstract: str = "",
    license_type: str = Document.License.CC_BY,
    license_notes: str = "",
    language: str = "en",
    year: int | None = None,
    grant_reference: str = "",
    authors: list[Author] | None = None,
    visibility: str = Document.Visibility.INHERIT,
    submitted_by=None,
    source_system: str = "manual",
    actor=None,
    request=None,
    allow_duplicate: bool = False,
    doi: str = "",
) -> Document:
    """Validate file + metadata, persist Document in DRAFT, dispatch follow-up.

    Deliberately **not** decorated with ``@transaction.atomic`` as a whole:
    ``_validate_repository_upload`` must run outside any wrapping transaction
    so that a FRREP-DCI007 malware-detection audit row (written via
    ``transaction.on_commit`` inside ``_report_malware_detection``) actually
    survives — a rejected upload always raises, and raising from inside an
    atomic block would roll back and silently discard that audit row. Once
    validation passes, everything else runs in its own atomic block as before.
    """
    _validate_repository_upload(file, actor=actor, request=request)
    file_hash = compute_file_hash(file)

    if not allow_duplicate:
        existing = detect_duplicate(file_hash)
        if existing is not None:
            raise DuplicateDocumentError(existing)

    return _ingest_document_atomic(
        file=file,
        file_hash=file_hash,
        title=title,
        collection=collection,
        document_type=document_type,
        abstract=abstract,
        license_type=license_type,
        license_notes=license_notes,
        language=language,
        year=year,
        grant_reference=grant_reference,
        authors=authors,
        visibility=visibility,
        submitted_by=submitted_by,
        source_system=source_system,
        actor=actor,
        request=request,
        doi=doi,
    )


@transaction.atomic
def _ingest_document_atomic(
    *,
    file,
    file_hash: str,
    title: str,
    collection: Collection,
    document_type: str,
    abstract: str,
    license_type: str,
    license_notes: str,
    language: str,
    year: int | None,
    grant_reference: str,
    authors: list[Author] | None,
    visibility: str,
    submitted_by,
    source_system: str,
    actor,
    request,
    doi: str,
) -> Document:
    document = Document(
        title=title.strip()[:400],
        abstract=abstract.strip(),
        file=file,
        file_hash=file_hash,
        file_size_bytes=getattr(file, "size", 0) or 0,
        document_type=document_type,
        doi=doi.strip(),
        license_type=license_type,
        license_notes=license_notes,
        language=language,
        year=year,
        collection=collection,
        visibility=visibility,
        grant_reference=grant_reference.strip(),
        submitted_by=submitted_by,
        source_system=source_system,
    )
    document.public_document_id = _allocate_public_document_id()
    document.save()

    for idx, author in enumerate(authors or []):
        DocumentAuthor.objects.get_or_create(
            document=document,
            author=author,
            defaults={
                "order": idx,
                "role": (
                    DocumentAuthor.AuthorRole.PRIMARY
                    if idx == 0
                    else DocumentAuthor.AuthorRole.CO_AUTHOR
                ),
            },
        )

    # FRREP-VL004 — every newly ingested document starts at version 1.0. Record
    # it as a DocumentVersion so the version history is populated from the outset
    # (rather than appearing only once a *second* version is uploaded). The
    # version points at the already-stored file by name so we don't re-copy the
    # bytes on disk.
    initial_version = DocumentVersion(
        document=document,
        version_number="1.0",
        file_hash=document.file_hash or "",
        uploaded_by=submitted_by,
        changelog="Initial version.",
    )
    initial_version.file.name = document.file.name
    initial_version.save()

    # FRREP-DCI017 — capture what was ingested, how, and the validation outcome
    # (not just "a Document was created") so the audit trail can answer "which
    # file was this, and did it pass the intake checks?" without a join.
    audit_repository(
        actor=actor,
        action=action_create(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={
            "filename": getattr(file, "name", "") or "",
            "ingestion_method": source_system,
            "file_size_bytes": document.file_size_bytes,
            "validation": "passed",
        },
        request=request,
    )

    # FRREP-DCI014 — fan out an "ingested" notification to the per-collection
    # subscriber list configured by the Repository Administrator, plus anyone
    # holding one of the collection's role-based subscriptions.
    recipients_by_id = {
        u.pk: u for u in collection.upload_notification_recipients.filter(is_active=True)
    }
    if collection.notify_roles:
        for u in User.objects.filter(role__in=collection.notify_roles, is_active=True):
            recipients_by_id.setdefault(u.pk, u)
    recipients = list(recipients_by_id.values())
    if recipients:
        from apps.core.notifications.services import bulk_notify

        bulk_notify(
            recipients,
            f'New document ingested into "{collection.name}": {document.title[:80]}.',
            verb=Notification.Verb.REPO_DOC_INGESTED,
        )

    # Async post-processing — text extraction, thumbnail, search index, AI prep.
    from apps.repository.documents.tasks import (
        extract_text_from_file,
        generate_thumbnail,
        index_document_for_search,
    )

    transaction.on_commit(lambda doc_id=document.pk: extract_text_from_file.delay(doc_id))
    transaction.on_commit(lambda doc_id=document.pk: generate_thumbnail.delay(doc_id))
    transaction.on_commit(lambda doc_id=document.pk: index_document_for_search.delay(doc_id))
    return document


# ── QA workflow ────────────────────────────────────────────────────────────


# SRS default mandatory metadata set (FRREP-DCI010 / MCL010) used whenever the
# target collection has no MetadataSchema of its own.
DEFAULT_MANDATORY_FIELDS = ["title", "abstract", "license_type"]


def mandatory_metadata_fields(collection: Collection | None) -> list[str]:
    """The metadata field codes a submission must fill for ``collection``.

    Driven by the collection's :class:`MetadataSchema` when one is configured,
    otherwise the SRS default set. This is the **single source of truth** shared
    by the submit form (which enforces it *before* ingest) and the QA pre-check
    (which re-checks it after submit) so the two cannot drift.
    """
    schema = collection.metadata_schema if collection and collection.metadata_schema_id else None
    if schema is not None:
        return list(schema.mandatory_fields)
    return list(DEFAULT_MANDATORY_FIELDS)


# FRREP-QA001 — a well-formed ORCID iD ("0000-0000-0000-000X"). Format-only:
# this is an offline pre-check gate, not a live registry lookup (that already
# happens separately in `register_author` / `verify_orcid`).
_ORCID_FORMAT_RE = re.compile(r"^\d{4}-\d{4}-\d{4}-\d{3}[\dX]$")

# Human-readable failure copy surfaced back to the submitter (FRREP-QA001) —
# keyed by the same codes `_run_pre_checks` writes into its notes dict.
_PRE_CHECK_MESSAGES: dict[str, str] = {
    "title": "Title is missing.",
    "abstract": "Abstract is missing.",
    "license_type": "License type is missing.",
    "has_author": "At least one author is required.",
    "author_identifiers": "One or more author ORCID iDs are not validly formatted "
    "(expected 0000-0000-0000-0000).",
}

# FRREP-QA001 — pre-check codes that hard-block `submit_for_qa` outright
# (see the docstring at its call site for why this set is deliberately
# narrow rather than every code in `_PRE_CHECK_MESSAGES`).
_BLOCKING_PRE_CHECK_CODES = {"author_identifiers"}


def _run_pre_checks(document: Document) -> tuple[bool, dict[str, Any]]:
    notes: dict[str, Any] = {}
    mandatory = mandatory_metadata_fields(
        document.collection if document.collection_id else None
    )
    for field_code in mandatory:
        value = getattr(document, field_code, None)
        if isinstance(value, str):
            ok = bool(value.strip())
        else:
            ok = value not in (None, "")
        notes[field_code] = "ok" if ok else "missing"
    notes["has_author"] = "ok" if document.authors.exists() else "missing"
    # FRREP-QA001 — author-identifier *validity*, not just presence: a
    # malformed ORCID string slipping through is worse than none at all since
    # it looks authoritative in citations/exports.
    bad_orcids = [
        a.orcid
        for a in document.authors.all()
        if a.orcid and not _ORCID_FORMAT_RE.match(a.orcid)
    ]
    notes["author_identifiers"] = "ok" if not bad_orcids else "invalid"
    passed = all(v == "ok" for v in notes.values())
    return passed, notes


def pre_check_failure_messages(notes: dict[str, Any]) -> list[str]:
    """Human-readable reasons for every failing ``_run_pre_checks`` code."""
    return [
        _PRE_CHECK_MESSAGES.get(code, f"Pre-check failed: {code}.")
        for code, outcome in notes.items()
        if outcome != "ok"
    ]


class PreCheckFailedError(ValidationError):
    """Raised by ``submit_for_qa`` when FRREP-QA001 pre-checks fail.

    Previously a failing pre-check was routed silently into the human QA
    queue — a reviewer would eventually reject it for something the submitter
    could have fixed immediately. This carries the specific failure list back
    to the caller instead, so ``DocumentSubmitView`` can surface it as form
    errors right away.
    """

    def __init__(self, failures: list[str]):
        super().__init__(failures)
        self.failures = list(failures)


def _select_qa_reviewer(document: Document):
    managed = document.collection.managed_by.filter(is_active=True)
    if managed.exists():
        return managed.first()
    manager = User.objects.filter(role=UserRole.REPO_MANAGER, is_active=True).first()
    if manager:
        return manager
    # SP6 two-tier review — dedicated Reviewer-role users are a valid
    # assignment pool when no collection manager / repo manager exists.
    return User.objects.filter(role=UserRole.REVIEWER, is_active=True).first()


@transaction.atomic
def submit_for_qa(document: Document, *, actor=None, request=None) -> Document:
    if document.status not in {Document.Status.DRAFT, Document.Status.REVISION_REQUIRED}:
        raise ValidationError("Document is not in a submittable state.")

    if document.status == Document.Status.REVISION_REQUIRED:
        document.resubmit()
    else:
        document.submit()
    document.save(update_fields=["status", "submitted_at", "updated_at"])

    passed, notes = _run_pre_checks(document)

    # FRREP-QA001 — a malformed author identifier is a brand-new check with no
    # existing funnel that could have caught it earlier, so it hard-blocks the
    # submission with a specific reason. The mandatory-metadata / has-author
    # checks below are already enforced at the DocumentSubmitForm level for
    # every interactive submission (FRREP-DCI010); when one still reaches this
    # point — bulk/API/legacy paths that bypass the form — the long-standing
    # behaviour of routing to the manual QA queue is preserved rather than
    # discarding the submission outright.
    blocking = [
        code for code in _BLOCKING_PRE_CHECK_CODES if notes.get(code) not in (None, "ok")
    ]
    if blocking:
        raise PreCheckFailedError(pre_check_failure_messages({c: notes[c] for c in blocking}))

    if not document.collection.qa_workflow_enabled and passed:
        # Auto-publish path (PRD REPO-SP6 — QA disabled).
        document.start_qa_review()
        document.save(update_fields=["status", "updated_at"])
        document.approve()
        document.save(update_fields=["status", "updated_at"])
        publish_document(document, actor=actor, request=request)
        return document

    document.start_qa_review()
    document.save(update_fields=["status", "updated_at"])

    reviewer = _select_qa_reviewer(document)
    qa = QARecord.objects.create(
        document=document,
        reviewer=reviewer,
        pre_check_passed=passed,
        pre_check_notes=notes,
        decision=QARecord.Decision.PENDING,
    )

    # A Reviewer-role assignee is not in REPOSITORY_MANAGEMENT_ROLES, so
    # restricted/internal submissions would be invisible to them — grant
    # per-object view/download for the document under review.
    if reviewer is not None:
        grant_object_perm(reviewer, "repository_documents.view_document", document)
        grant_object_perm(reviewer, "repository_documents.download_document", document)

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"status": {"old": "draft", "new": document.status}},
        request=request,
    )

    repo_signals.document_qa_assigned.send(
        sender=Document,
        document=document,
        qa_record=qa,
    )
    repo_signals.document_submitted.send(sender=Document, document=document)
    return document


def full_qa_checklist() -> dict:
    """All FRREP-QA003 checklist criteria marked true — convenience for
    programmatic approvals (seeders, tests, integrations) that bypass the
    review form but still need to satisfy the approval gate."""
    return {code: True for code, _ in QARecord.CHECKLIST_CRITERIA}


@transaction.atomic
def decide_qa(
    qa_record: QARecord,
    *,
    decision: str,
    comments: str = "",
    checklist: dict | None = None,
    actor=None,
    request=None,
) -> Document:
    if qa_record.decision != QARecord.Decision.PENDING:
        raise ValidationError("This QA record has already been decided.")
    document = qa_record.document
    if document.status != Document.Status.UNDER_QA:
        raise ValidationError("Document is not currently under QA review.")

    # SP6 two-tier review — when a finalizer doesn't pass a checklist (e.g. a
    # programmatic caller), fall back to the reviewer's persisted rubric rather
    # than silently rebuilding it as all-False. The interactive finalize form
    # always resends the checklist (pre-filled from the recommendation via
    # QAReviewView.get_initial), so this only affects non-view callers.
    if checklist is None:
        checklist = dict(qa_record.checklist or {})
    checklist = {
        code: bool(checklist.get(code)) for code, _ in QARecord.CHECKLIST_CRITERIA
    }
    # FRREP-QA003 — server-side gate: approval requires every checklist
    # criterion to be explicitly checked. Revision/reject stay unblocked so a
    # reviewer can reject *because* e.g. copyright is unchecked.
    if decision == QARecord.Decision.APPROVE and not all(checklist.values()):
        raise ValidationError(
            "All 6 QA checklist items must be checked before a document can be approved."
        )

    qa_record.decision = decision
    qa_record.comments = comments
    qa_record.checklist = checklist
    qa_record.decided_at = timezone.now()
    qa_record.save(update_fields=["decision", "comments", "checklist", "decided_at"])

    # SP6 least-privilege — the per-object view/download grant made to the
    # assigned reviewer in ``submit_for_qa`` was scoped to the review task.
    # Revoke it now the QA lifecycle has ended so a pure REVIEWER doesn't retain
    # standing access to a (possibly restricted) document — most importantly
    # after a reject/withdraw, where nothing else re-establishes access. On the
    # APPROVE path, ``apply_collection_visibility`` (via ``publish_document``
    # below) re-grants whoever should keep access — collection managers and the
    # submitter — so a reviewer who is also a collection manager is unaffected.
    # On resubmit-after-revision, ``submit_for_qa`` re-grants the next reviewer.
    if qa_record.reviewer_id:
        revoke_object_perm(qa_record.reviewer, "repository_documents.view_document", document)
        revoke_object_perm(qa_record.reviewer, "repository_documents.download_document", document)

    if decision == QARecord.Decision.APPROVE:
        document.approve()
        document.save(update_fields=["status", "updated_at"])
        repo_signals.document_approved.send(sender=Document, document=document)
        publish_document(document, actor=actor, request=request)
        return document

    if decision == QARecord.Decision.REVISION:
        if not comments.strip():
            raise ValidationError("Comments are required when requesting revisions.")
        document.request_revision()
        document.save(update_fields=["status", "updated_at"])
        repo_signals.document_qa_revision_requested.send(
            sender=Document,
            document=document,
            qa_record=qa_record,
        )
    elif decision == QARecord.Decision.REJECT:
        if not comments.strip():
            raise ValidationError("Comments are required when rejecting a document.")
        document.reject()
        document.save(update_fields=["status", "updated_at"])
        repo_signals.document_rejected.send(sender=Document, document=document, qa_record=qa_record)
    else:
        raise ValidationError(f"Unknown QA decision: {decision}")

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"qa_decision": decision},
        request=request,
    )

    if document.submitted_by_id:
        # FRREP-QA006 — the reviewer's actual comments must reach the
        # submitter; a generic status-change line leaves them guessing what
        # to fix before resubmitting.
        message = f'Your submission "{document.title[:80]}" was {document.get_status_display()}.'
        if comments.strip():
            message += f' Reviewer comments: "{comments.strip()[:300]}"'
        send_notification(
            document.submitted_by,
            message,
            verb=(
                Notification.Verb.REPO_DOC_REVISION_REQUIRED
                if decision == QARecord.Decision.REVISION
                else Notification.Verb.REPO_DOC_REJECTED
            ),
        )
    return document


@transaction.atomic
def recommend_qa(
    qa_record: QARecord,
    *,
    recommendation: str,
    comments: str = "",
    checklist: dict | None = None,
    actor=None,
    request=None,
) -> QARecord:
    """SP6 two-tier review — a Reviewer scores the rubric and records a
    recommendation; the final decision stays with ``decide_qa`` (management
    tier). Re-recommending overwrites the previous recommendation while the
    record is still pending."""
    if qa_record.decision != QARecord.Decision.PENDING:
        raise ValidationError("This QA record has already been decided.")
    if qa_record.document.status != Document.Status.UNDER_QA:
        raise ValidationError("Document is not currently under QA review.")
    if recommendation not in {
        QARecord.Decision.APPROVE,
        QARecord.Decision.REVISION,
        QARecord.Decision.REJECT,
    }:
        raise ValidationError(f"Unknown QA recommendation: {recommendation}")
    if recommendation != QARecord.Decision.APPROVE and not comments.strip():
        raise ValidationError("Comments are required when recommending revision or rejection.")

    checklist = {
        code: bool((checklist or {}).get(code)) for code, _ in QARecord.CHECKLIST_CRITERIA
    }
    if recommendation == QARecord.Decision.APPROVE and not all(checklist.values()):
        raise ValidationError(
            "All 6 QA checklist items must be checked to recommend approval."
        )

    qa_record.recommendation = recommendation
    qa_record.recommendation_comments = comments
    qa_record.recommended_by = actor
    qa_record.recommended_at = timezone.now()
    qa_record.checklist = checklist
    qa_record.save(
        update_fields=[
            "recommendation",
            "recommendation_comments",
            "recommended_by",
            "recommended_at",
            "checklist",
        ]
    )

    document = qa_record.document
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"qa_recommendation": recommendation},
        request=request,
    )

    # The recommendation is only useful if a finalizer hears about it —
    # Editors-in-Chief and repo managers (both hold finalize authority) plus
    # this collection's managers. Repo managers are included so that when the
    # assigned reviewer/actor is the collection's *only* manager and no EIC
    # exists, the recommendation still reaches someone who can act on it.
    from apps.core.notifications.services import bulk_notify

    finalizers = list(
        User.objects.filter(
            role__in=[UserRole.EDITOR_IN_CHIEF, UserRole.REPO_MANAGER], is_active=True
        )
    )
    for manager in document.collection.managed_by.filter(is_active=True):
        if manager not in finalizers:
            finalizers.append(manager)
    finalizers = [u for u in finalizers if u != actor]
    if finalizers:
        bulk_notify(
            finalizers,
            f'Reviewer recommendation on "{document.title[:80]}": '
            f"{qa_record.get_recommendation_display()} — awaiting final decision.",
            verb=Notification.Verb.REPO_QA_RECOMMENDED,
        )
    return qa_record


@transaction.atomic
def record_compliance_check(
    qa_record: QARecord,
    *,
    status: str,
    notes: str = "",
    actor=None,
    request=None,
) -> QARecord:
    """SP6 — Publication Assistant's licensing / open-access compliance check.

    Advisory to the final QA decision: it never blocks ``decide_qa``, but the
    state is surfaced on the queue and review pages, and a FLAGGED result
    notifies the collection's managers."""
    if qa_record.decision != QARecord.Decision.PENDING:
        raise ValidationError("This QA record has already been decided.")
    # Mirror recommend_qa's guard — a document can leave UNDER_QA (e.g. a
    # ``source="*"`` withdraw) while its QARecord.decision is still PENDING, and
    # a compliance check on a no-longer-under-review document is meaningless.
    if qa_record.document.status != Document.Status.UNDER_QA:
        raise ValidationError("Document is not currently under QA review.")
    if status not in {
        QARecord.ComplianceStatus.PASSED,
        QARecord.ComplianceStatus.FLAGGED,
    }:
        raise ValidationError(f"Unknown compliance status: {status}")
    if status == QARecord.ComplianceStatus.FLAGGED and not notes.strip():
        raise ValidationError("Notes are required when flagging a compliance issue.")

    qa_record.compliance_status = status
    qa_record.compliance_notes = notes
    qa_record.compliance_checked_by = actor
    qa_record.compliance_checked_at = timezone.now()
    qa_record.save(
        update_fields=[
            "compliance_status",
            "compliance_notes",
            "compliance_checked_by",
            "compliance_checked_at",
        ]
    )

    document = qa_record.document
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"qa_compliance": status},
        request=request,
    )

    if status == QARecord.ComplianceStatus.FLAGGED:
        from apps.core.notifications.services import bulk_notify

        recipients = [
            u
            for u in document.collection.managed_by.filter(is_active=True)
            if u != actor
        ]
        if qa_record.reviewer and qa_record.reviewer != actor and qa_record.reviewer not in recipients:
            recipients.append(qa_record.reviewer)
        if recipients:
            bulk_notify(
                recipients,
                f'Compliance check flagged "{document.title[:80]}": {notes.strip()[:200]}',
                verb=Notification.Verb.REPO_QA_COMPLIANCE_FLAGGED,
            )
    return qa_record


def log_access_violation(request, document: Document, *, action: str) -> None:
    """REPO-SP4 — persist a denied access/download attempt so Repository
    Administrators have a reviewable trail (previously denials only produced
    a 403). Never raises: a logging failure must not mask the 403 itself."""
    from apps.repository.documents.models import AccessViolation

    try:
        user = getattr(request, "user", None)
        AccessViolation.objects.create(
            document=document,
            user=user if getattr(user, "is_authenticated", False) else None,
            action=action,
            ip_address=(request.META.get("REMOTE_ADDR") or None),
            user_agent=request.META.get("HTTP_USER_AGENT", "")[:512],
            path=request.get_full_path()[:500],
        )
    except Exception:  # pragma: no cover — best-effort logging only
        logger.exception("failed to record access violation for doc %s", document.pk)


@transaction.atomic
def publish_document(document: Document, *, actor=None, request=None) -> Document:
    if document.status != Document.Status.APPROVED:
        raise ValidationError("Only APPROVED documents can be published.")
    document.publish()
    document.save(update_fields=["status", "published_at", "updated_at"])

    from apps.repository.documents.permissions import apply_collection_visibility

    apply_collection_visibility(document)

    # Build the FTS search_vector synchronously so the document is searchable
    # immediately, even if the Celery worker is offline. Celery still re-runs
    # this asynchronously via the post_save receiver — the work is idempotent.
    from apps.repository.search.indexes import update_search_vector

    transaction.on_commit(lambda doc_id=document.pk: update_search_vector(doc_id))

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"status": {"old": "approved", "new": "published"}},
        request=request,
    )

    if document.submitted_by_id:
        send_notification(
            document.submitted_by,
            f'Your submission "{document.title[:80]}" has been published in the Repository.',
            verb=Notification.Verb.REPO_DOC_PUBLISHED,
        )

    repo_signals.document_published.send(sender=Document, document=document)
    return document


@transaction.atomic
def withdraw_document(
    document: Document,
    *,
    reason: str,
    actor=None,
    request=None,
) -> Document:
    if not reason.strip():
        raise ValidationError("A reason is required to withdraw a document.")
    document.withdraw()
    document.withdrawn_reason = reason.strip()
    document.save(update_fields=["status", "withdrawn_at", "withdrawn_reason", "updated_at"])

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"status": "withdrawn", "reason": reason},
        request=request,
    )
    repo_signals.document_withdrawn.send(sender=Document, document=document, reason=reason)
    return document


@transaction.atomic
def reinstate_document(
    document: Document,
    *,
    reason: str = "",
    actor=None,
    request=None,
) -> Document:
    document.reinstate()
    document.save(update_fields=["status", "withdrawn_at", "updated_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"status": "approved", "reason": reason},
        request=request,
    )
    repo_signals.document_reinstated.send(sender=Document, document=document)
    return document


# ── access requests / sharing ───────────────────────────────────────────────


@transaction.atomic
def request_access(document: Document, requester, justification: str) -> AccessRequest:
    if not justification.strip():
        raise ValidationError("Please describe why you need access.")
    ar = AccessRequest.objects.create(
        document=document,
        requester=requester,
        justification=justification.strip(),
    )
    repo_signals.access_requested.send(sender=Document, document=document, access_request=ar)
    return ar


@transaction.atomic
def decide_access_request(
    access_request: AccessRequest,
    *,
    approve: bool,
    note: str = "",
    actor,
    request=None,
) -> AccessRequest:
    if access_request.status != AccessRequest.Status.PENDING:
        raise ValidationError("Already decided.")
    access_request.status = (
        AccessRequest.Status.APPROVED if approve else AccessRequest.Status.DENIED
    )
    access_request.decided_by = actor
    access_request.decided_at = timezone.now()
    access_request.decision_note = note
    access_request.save(
        update_fields=["status", "decided_by", "decided_at", "decision_note"]
    )

    if approve:
        grant_object_perm(
            access_request.requester,
            "repository_documents.download_document",
            access_request.document,
        )
        grant_object_perm(
            access_request.requester,
            "repository_documents.view_document",
            access_request.document,
        )
        send_notification(
            access_request.requester,
            f'Access granted to "{access_request.document.title[:80]}".',
            verb=Notification.Verb.REPO_ACCESS_GRANTED,
        )
        repo_signals.access_granted.send(
            sender=Document,
            document=access_request.document,
            access_request=access_request,
        )
    else:
        repo_signals.access_denied.send(
            sender=Document,
            document=access_request.document,
            access_request=access_request,
        )

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="AccessRequest",
        object_id=access_request.pk,
        object_repr=str(access_request.document),
        changes={"status": access_request.status},
        request=request,
    )
    return access_request


_PERMISSION_LEVEL_MAP = {
    DocumentShare.PermissionLevel.VIEW: ["repository_documents.view_document"],
    DocumentShare.PermissionLevel.COMMENT: [
        "repository_documents.view_document",
        "repository_documents.comment_document",
    ],
    DocumentShare.PermissionLevel.EDIT: [
        "repository_documents.view_document",
        "repository_documents.comment_document",
        "repository_documents.change_document",
    ],
}


@transaction.atomic
def share_document(
    document: Document,
    collaborator,
    permission_level: str,
    *,
    granted_by,
    request=None,
) -> DocumentShare:
    share, _ = DocumentShare.objects.update_or_create(
        document=document,
        collaborator=collaborator,
        defaults={
            "permission_level": permission_level,
            "granted_by": granted_by,
            "revoked_at": None,
        },
    )
    for perm in _PERMISSION_LEVEL_MAP.get(permission_level, []):
        grant_object_perm(collaborator, perm, document)
    audit_repository(
        actor=granted_by,
        action=action_create(),
        target_model="DocumentShare",
        object_id=share.pk,
        object_repr=str(document),
        changes={"collaborator_id": collaborator.pk, "permission_level": permission_level},
        request=request,
    )
    # FRREP-COL002 — share recipients get a notification with the permission level.
    granter_name = granted_by.get_full_name() or granted_by.email if granted_by else "A collaborator"
    send_notification(
        collaborator,
        f'{granter_name} shared "{document.title[:80]}" with you ({permission_level}).',
        verb=Notification.Verb.REPO_DOC_SHARED,
    )
    return share


@transaction.atomic
def revoke_share(share: DocumentShare, *, actor, request=None) -> DocumentShare:
    for perm in _PERMISSION_LEVEL_MAP.get(share.permission_level, []):
        revoke_object_perm(share.collaborator, perm, share.document)
    share.revoked_at = timezone.now()
    share.save(update_fields=["revoked_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="DocumentShare",
        object_id=share.pk,
        object_repr=str(share.document),
        changes={"revoked_at": share.revoked_at.isoformat()},
        request=request,
    )
    return share


# ── group-based collection access (FRREP-AC005) ─────────────────────────────


@transaction.atomic
def grant_collection_access(
    collection: Collection,
    group,
    *,
    permission_level: str = CollectionGroupAccess.PermissionLevel.VIEW,
    actor=None,
    request=None,
    expires_at=None,
    note: str = "",
) -> CollectionGroupAccess:
    """Grant or refresh access for `group` to `collection`.

    If a grant already exists it is updated in place — `granted_at` is preserved
    and the `expires_at` / `permission_level` / `note` are overwritten. Audit
    captures both the create and the update path.
    """
    if permission_level not in dict(CollectionGroupAccess.PermissionLevel.choices):
        raise ValidationError("Unknown permission level.")
    existing = CollectionGroupAccess.objects.filter(
        collection=collection, group=group
    ).first()
    created = existing is None
    if existing is None:
        grant = CollectionGroupAccess.objects.create(
            collection=collection,
            group=group,
            permission_level=permission_level,
            granted_by=actor,
            expires_at=expires_at,
            note=note.strip(),
        )
    else:
        existing.permission_level = permission_level
        existing.granted_by = actor or existing.granted_by
        existing.expires_at = expires_at
        existing.note = note.strip()
        existing.save(
            update_fields=["permission_level", "granted_by", "expires_at", "note"]
        )
        grant = existing
    audit_repository(
        actor=actor,
        action=action_create() if created else action_update(),
        target_model="CollectionGroupAccess",
        object_id=grant.pk,
        object_repr=str(grant),
        changes={
            "collection_id": collection.pk,
            "group_id": group.pk,
            "permission_level": permission_level,
            "expires_at": expires_at.isoformat() if expires_at else None,
        },
        request=request,
    )
    return grant


@transaction.atomic
def revoke_collection_access(
    grant: CollectionGroupAccess,
    *,
    actor=None,
    request=None,
) -> None:
    """Hard-delete a CollectionGroupAccess grant.

    Audit row preserves who, what, and the resolved grant string so the
    historical context survives the delete.
    """
    audit_repository(
        actor=actor,
        action=action_delete(),
        target_model="CollectionGroupAccess",
        object_id=grant.pk,
        object_repr=str(grant),
        changes={
            "collection_id": grant.collection_id,
            "group_id": grant.group_id,
            "permission_level": grant.permission_level,
        },
        request=request,
    )
    grant.delete()


# ── Author / taxonomy merge wizards (FRREP-MCL004 + FRREP-MCL012) ──────────


@transaction.atomic
def merge_authors(
    primary: Author,
    duplicates: list[Author],
    *,
    actor=None,
    request=None,
) -> Author:
    """Soft-merge ``duplicates`` into ``primary`` (FRREP-MCL004).

    Each duplicate's display name is snapshotted onto the primary as an
    ``AuthorAlias(source=MERGE)`` so historical bylines remain discoverable.
    The duplicate row itself is **not deleted** — it is archived in place
    (``is_archived=True``, ``merged_into=primary``) so external systems that
    referenced the duplicate by id can still resolve it via the FK chain.

    Edge cases handled:

    * If the primary already authored a document one of the duplicates also
      authored, the duplicate's DocumentAuthor row is dropped (uniqueness
      constraint on ``(document, author)``) — the primary's existing row wins,
      preserving its ``order`` and ``role``.
    * `affiliation_text` and `email` from duplicates are appended to the
      primary's free-text fields when the primary's value is empty — never
      overwritten.
    * The primary's ``orcid`` and ``verified_at`` are always preserved.
    * Re-merging an already-archived author is a no-op for that row.
    * A single audit row records the rewire with before/after author IDs and
      the alias count.
    """
    if any(d.pk == primary.pk for d in duplicates):
        raise ValidationError("A primary author cannot be merged into itself.")
    if not duplicates:
        raise ValidationError("Provide at least one duplicate to merge.")

    duplicate_ids = [d.pk for d in duplicates]
    primary_doc_ids = set(
        DocumentAuthor.objects.filter(author=primary).values_list("document_id", flat=True)
    )

    aliases_created = 0
    for d in duplicates:
        name = d.full_name
        if not name:
            continue
        _, created = AuthorAlias.objects.get_or_create(
            author=primary,
            full_name=name,
            defaults={
                "source": AuthorAlias.Source.MERGE,
                "note": f"Merged from author #{d.pk} on {timezone.now().date().isoformat()}",
            },
        )
        if created:
            aliases_created += 1

    rewired = 0
    dropped_due_to_collision = 0

    for da in DocumentAuthor.objects.filter(author_id__in=duplicate_ids):
        if da.document_id in primary_doc_ids:
            da.delete()
            dropped_due_to_collision += 1
        else:
            da.author = primary
            da.save(update_fields=["author"])
            primary_doc_ids.add(da.document_id)
            rewired += 1

    dirty: list[str] = []
    if not primary.email:
        for d in duplicates:
            if d.email:
                primary.email = d.email
                dirty.append("email")
                break
    if not primary.affiliation_text:
        merged = "; ".join(
            d.affiliation_text for d in duplicates if d.affiliation_text
        )
        if merged:
            primary.affiliation_text = merged[:255]
            dirty.append("affiliation_text")
    if dirty:
        primary.save(update_fields=dirty + ["updated_at"])

    # Soft-delete: archive duplicate rows + point at primary. Leaves the row in
    # the DB so historical FKs and audit queries still resolve.
    Author.all_objects.filter(pk__in=duplicate_ids).update(
        is_archived=True,
        merged_into=primary,
        updated_at=timezone.now(),
    )

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Author",
        object_id=primary.pk,
        object_repr=str(primary),
        changes={
            "merged_from_author_ids": duplicate_ids,
            "rewired_document_authors": rewired,
            "dropped_collisions": dropped_due_to_collision,
            "fields_backfilled": dirty,
            "aliases_created": aliases_created,
        },
        request=request,
    )
    return primary


@transaction.atomic
def rename_author(
    author: Author,
    *,
    first_name: str,
    last_name: str,
    actor=None,
    request=None,
) -> Author:
    """Rename an author in place, snapshotting the previous name as a
    RENAME-source ``AuthorAlias`` (FRREP-MCL004 name-change linking).

    No-op when the supplied first + last name match the current values.
    """
    new_first = (first_name or "").strip()
    new_last = (last_name or "").strip()
    if not (new_first or new_last):
        raise ValidationError("Provide at least one of first_name / last_name.")

    if new_first == author.first_name and new_last == author.last_name:
        return author

    previous_name = author.full_name
    if previous_name:
        AuthorAlias.objects.get_or_create(
            author=author,
            full_name=previous_name,
            defaults={
                "source": AuthorAlias.Source.RENAME,
                "note": f"Renamed on {timezone.now().date().isoformat()}",
            },
        )

    author.first_name = new_first or author.first_name
    author.last_name = new_last or author.last_name
    author.save(update_fields=["first_name", "last_name", "updated_at"])

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Author",
        object_id=author.pk,
        object_repr=str(author),
        changes={
            "previous_name": previous_name,
            "new_first_name": new_first,
            "new_last_name": new_last,
        },
        request=request,
    )
    return author


@transaction.atomic
def merge_collections(
    primary: Collection,
    duplicates: list[Collection],
    *,
    actor=None,
    request=None,
) -> Collection:
    """Move every Document from ``duplicates`` to ``primary``, write slug
    aliases so old URLs still 301-resolve, then **deprecate** each duplicate
    in place rather than deleting it (FRREP-MCL012).

    Deprecation preserves the historical row, lets ``superseded_by`` resolve
    queries, and keeps the slug alias chain intact for audit. The duplicate
    is no longer offered to users for new submissions because the SubmitForm
    refuses deprecated collections without a fallback.

    Refuses the merge when:

    * The primary appears in the duplicates list.
    * Any duplicate has a different ``visibility`` from the primary — merging a
      Public collection into a Restricted one (or vice-versa) silently changes
      the access posture of every moved document, which is too dangerous to do
      without an explicit visibility-change step first.
    """
    if any(d.pk == primary.pk for d in duplicates):
        raise IncompatibleMergeError(
            "A collection cannot be merged into itself."
        )
    if not duplicates:
        raise ValidationError("Provide at least one duplicate collection.")
    bad_visibility = [d for d in duplicates if d.visibility != primary.visibility]
    if bad_visibility:
        names = ", ".join(d.name for d in bad_visibility)
        raise IncompatibleMergeError(
            f"Visibility mismatch: {names} differs from {primary.name} "
            f"({primary.get_visibility_display()}). Align visibility on each "
            "collection before merging."
        )

    duplicate_ids = [d.pk for d in duplicates]
    duplicate_slugs = [d.slug for d in duplicates]

    moved = Document.objects.filter(collection_id__in=duplicate_ids).update(
        collection=primary
    )

    # Rewire children + group-access grants + email config off the duplicates.
    Collection.objects.filter(parent_id__in=duplicate_ids).update(parent=primary)

    # Preserve old slugs as 301-redirect aliases.
    for slug in duplicate_slugs:
        if not CollectionSlugAlias.objects.filter(old_slug=slug).exists():
            CollectionSlugAlias.objects.create(
                collection=primary,
                old_slug=slug,
                note=f"Merged from collection '{slug}' on {timezone.now().date().isoformat()}",
            )

    # Deprecate-in-place instead of delete. Disable email ingestion so the
    # deprecated row no longer competes for inbound mail.
    today = timezone.now().date().isoformat()
    for dup in duplicates:
        dup.is_deprecated = True
        dup.deprecation_message = f"Merged into '{primary.name}' on {today}."
        dup.deprecated_at = timezone.now()
        dup.superseded_by = primary
        if dup.email_ingest_enabled:
            dup.email_ingest_enabled = False
        dup.save(update_fields=[
            "is_deprecated",
            "deprecation_message",
            "deprecated_at",
            "superseded_by",
            "email_ingest_enabled",
            "updated_at",
        ])

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Collection",
        object_id=primary.pk,
        object_repr=str(primary),
        changes={
            "merged_from_collection_ids": duplicate_ids,
            "merged_from_slugs": duplicate_slugs,
            "documents_moved": moved,
            "deprecated_with_supersession": True,
        },
        request=request,
    )
    return primary


@transaction.atomic
def deprecate_collection(
    collection: Collection,
    *,
    message: str,
    superseded_by: Collection | None = None,
    actor=None,
    request=None,
) -> Collection:
    """Mark a Collection as deprecated (FRREP-MCL012, SRS use case A4).

    The collection remains queryable so historical traffic is unaffected, but
    submission forms and listings render a warning. ``superseded_by`` (when
    provided) is surfaced as the recommended replacement.

    Idempotent — re-deprecating refreshes the message + supersession but
    leaves the original ``deprecated_at`` timestamp.
    """
    msg = (message or "").strip()
    if not msg:
        raise ValidationError("Deprecation message is required.")
    if superseded_by is not None and superseded_by.pk == collection.pk:
        raise ValidationError("A collection cannot supersede itself.")

    collection.is_deprecated = True
    collection.deprecation_message = msg[:255]
    if collection.deprecated_at is None:
        collection.deprecated_at = timezone.now()
    collection.superseded_by = superseded_by
    collection.save(update_fields=[
        "is_deprecated",
        "deprecation_message",
        "deprecated_at",
        "superseded_by",
        "updated_at",
    ])

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Collection",
        object_id=collection.pk,
        object_repr=str(collection),
        changes={
            "deprecated": True,
            "deprecation_message": msg,
            "superseded_by_id": superseded_by.pk if superseded_by else None,
        },
        request=request,
    )
    return collection


@transaction.atomic
def undeprecate_collection(
    collection: Collection,
    *,
    actor=None,
    request=None,
) -> Collection:
    """Reverse a deprecation. Clears the supersession and timestamp."""
    if not collection.is_deprecated:
        return collection
    collection.is_deprecated = False
    collection.deprecation_message = ""
    collection.deprecated_at = None
    collection.superseded_by = None
    collection.save(update_fields=[
        "is_deprecated",
        "deprecation_message",
        "deprecated_at",
        "superseded_by",
        "updated_at",
    ])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Collection",
        object_id=collection.pk,
        object_repr=str(collection),
        changes={"undeprecated": True},
        request=request,
    )
    return collection


@transaction.atomic
def rename_collection_slug(
    collection: Collection,
    new_slug: str,
    *,
    actor=None,
    request=None,
) -> Collection:
    """Change a collection's slug, recording the old slug as a redirect alias.

    Rejects the rename if `new_slug` collides with another collection's slug or
    any existing alias.
    """
    new_slug = (new_slug or "").strip().lower()
    if not new_slug:
        raise ValidationError("Slug cannot be empty.")
    if new_slug == collection.slug:
        return collection
    if Collection.objects.exclude(pk=collection.pk).filter(slug=new_slug).exists():
        raise ValidationError(
            f"Slug '{new_slug}' is already used by another collection."
        )
    if CollectionSlugAlias.objects.filter(old_slug=new_slug).exists():
        raise ValidationError(
            f"Slug '{new_slug}' is reserved by an existing alias and would "
            "create a redirect loop."
        )

    old_slug = collection.slug
    CollectionSlugAlias.objects.get_or_create(
        old_slug=old_slug,
        defaults={
            "collection": collection,
            "note": f"Renamed from '{old_slug}' on {timezone.now().date().isoformat()}",
        },
    )
    collection.slug = new_slug
    collection.save(update_fields=["slug", "updated_at"])

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Collection",
        object_id=collection.pk,
        object_repr=str(collection),
        changes={"slug": [old_slug, new_slug]},
        request=request,
    )
    return collection


# ── comments + review tasks ─────────────────────────────────────────────────


@transaction.atomic
def add_comment(
    document: Document,
    author,
    text: str,
    section_anchor: str = "",
    parent: DocumentComment | None = None,
    request=None,
) -> DocumentComment:
    if not text.strip():
        raise ValidationError("Comment text is required.")
    comment = DocumentComment.objects.create(
        document=document,
        author=author,
        text=text.strip(),
        section_anchor=section_anchor.strip(),
        parent=parent,
    )
    audit_repository(
        actor=author,
        action=action_create(),
        target_model="DocumentComment",
        object_id=comment.pk,
        object_repr=str(document),
        changes={"section_anchor": comment.section_anchor, "parent_id": parent.pk if parent else None},
        request=request,
    )
    return comment


@transaction.atomic
def resolve_comment(comment: DocumentComment, *, actor, request=None) -> DocumentComment:
    """Mark a comment thread resolved (FRREP-COL004). Notifies the original
    author when someone else closes it out."""
    if comment.is_resolved:
        return comment
    comment.is_resolved = True
    comment.save(update_fields=["is_resolved", "updated_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="DocumentComment",
        object_id=comment.pk,
        object_repr=str(comment.document),
        changes={"is_resolved": {"old": False, "new": True}},
        request=request,
    )
    if comment.author_id and actor and comment.author_id != actor.pk:
        actor_name = actor.get_full_name() or actor.email
        send_notification(
            comment.author,
            f'{actor_name} resolved your comment on "{comment.document.title[:80]}".',
            verb=Notification.Verb.REPO_COMMENT_RESOLVED,
        )
    return comment


@transaction.atomic
def reopen_comment(comment: DocumentComment, *, actor, request=None) -> DocumentComment:
    """Reverse ``resolve_comment`` — reopens the thread for further discussion."""
    if not comment.is_resolved:
        return comment
    comment.is_resolved = False
    comment.save(update_fields=["is_resolved", "updated_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="DocumentComment",
        object_id=comment.pk,
        object_repr=str(comment.document),
        changes={"is_resolved": {"old": True, "new": False}},
        request=request,
    )
    return comment


@transaction.atomic
def add_review_task(
    document: Document,
    *,
    assignee,
    scope: str,
    instructions: str = "",
    due_date=None,
    created_by,
    request=None,
) -> DocumentReviewTask:
    if not scope.strip():
        raise ValidationError("Review-task scope is required.")
    task = DocumentReviewTask.objects.create(
        document=document,
        assignee=assignee,
        scope=scope.strip(),
        instructions=instructions.strip(),
        due_date=due_date,
        created_by=created_by,
    )
    audit_repository(
        actor=created_by,
        action=action_create(),
        target_model="DocumentReviewTask",
        object_id=task.pk,
        object_repr=str(document),
        changes={
            "assignee_id": assignee.pk,
            "due_date": due_date.isoformat() if due_date else None,
        },
        request=request,
    )
    send_notification(
        assignee,
        f'You have been assigned a review task on "{document.title[:80]}".',
        verb=Notification.Verb.REPO_REVIEW_TASK_ASSIGNED,
    )
    return task


@transaction.atomic
def update_review_task_status(
    task: DocumentReviewTask, *, status: str, actor, request=None
) -> DocumentReviewTask:
    """Transition a review task's status (FRREP-COL005).

    Closes the loop `ReviewTaskCreateView` opened: without this, a task could
    be assigned but never marked in-progress/completed by the assignee. On
    completion, both the task's creator and the document's owner are notified
    (skipping whichever of them is the actor, so nobody is told about their
    own action).
    """
    valid_statuses = {choice for choice, _ in DocumentReviewTask.Status.choices}
    if status not in valid_statuses:
        raise ValidationError("Invalid review-task status.")
    if task.status == status:
        return task
    old_status = task.status
    task.status = status
    task.save(update_fields=["status"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="DocumentReviewTask",
        object_id=task.pk,
        object_repr=str(task.document),
        changes={"status": {"old": old_status, "new": status}},
        request=request,
    )
    if status == DocumentReviewTask.Status.COMPLETED:
        actor_name = actor.get_full_name() or actor.email if actor else "Someone"
        recipients = {}
        if task.created_by_id and (not actor or task.created_by_id != actor.pk):
            recipients[task.created_by_id] = task.created_by
        if task.document.submitted_by_id and (
            not actor or task.document.submitted_by_id != actor.pk
        ):
            recipients[task.document.submitted_by_id] = task.document.submitted_by
        for recipient in recipients.values():
            send_notification(
                recipient,
                f'{actor_name} marked the review task "{task.scope[:60]}" complete '
                f'on "{task.document.title[:60]}".',
                verb=Notification.Verb.REPO_REVIEW_TASK_COMPLETED,
            )
    return task


# ── versions ────────────────────────────────────────────────────────────────


def _next_version_number(document: Document, *, version_type: str = "major") -> str:
    """Compute the next version string (FRREP-VL002).

    ``version_type="major"`` bumps the integer part and resets the decimal to
    0 (e.g. 1.2 -> 2.0) — a substantive revision. ``version_type="minor"``
    keeps the integer part and increments the decimal (e.g. 1.0 -> 1.1 -> 1.2)
    — a small correction (typo/formatting). With no prior version, both start
    at 1.0 since there is nothing yet to correct.
    """
    latest = document.versions.order_by("-uploaded_at").first()
    if not latest:
        return "1.0"
    try:
        major_s, minor_s = latest.version_number.split(".", 1)
        major, minor = int(major_s), int(minor_s)
    except (ValueError, AttributeError):
        return f"{document.versions.count() + 1}.0"
    if version_type == "minor":
        return f"{major}.{minor + 1}"
    return f"{major + 1}.0"


def add_version(
    document: Document,
    *,
    file,
    changelog: str,
    uploaded_by,
    version_type: str = "major",
    request=None,
) -> DocumentVersion:
    # Not wrapped in @transaction.atomic as a whole — see ingest_document's
    # docstring: validation must run outside any atomic block so a
    # FRREP-DCI007 malware-audit row survives the raised ValidationError.
    if version_type not in {"major", "minor"}:
        raise ValidationError("version_type must be 'major' or 'minor'.")
    _validate_repository_upload(file, actor=uploaded_by, request=request)
    file_hash = compute_file_hash(file)
    return _add_version_atomic(
        document,
        file=file,
        file_hash=file_hash,
        changelog=changelog,
        uploaded_by=uploaded_by,
        version_type=version_type,
        request=request,
    )


@transaction.atomic
def _add_version_atomic(
    document: Document,
    *,
    file,
    file_hash: str,
    changelog: str,
    uploaded_by,
    version_type: str,
    request=None,
) -> DocumentVersion:
    version = DocumentVersion.objects.create(
        document=document,
        version_number=_next_version_number(document, version_type=version_type),
        file=file,
        file_hash=file_hash,
        uploaded_by=uploaded_by,
        changelog=changelog,
    )
    document.file = file
    document.file_hash = file_hash
    document.file_size_bytes = getattr(file, "size", 0) or 0
    document.save(update_fields=["file", "file_hash", "file_size_bytes", "updated_at"])

    audit_repository(
        actor=uploaded_by,
        action=action_update(),
        target_model="DocumentVersion",
        object_id=version.pk,
        object_repr=str(document),
        changes={"version": version.version_number, "version_type": version_type},
        request=request,
    )
    repo_signals.version_added.send(sender=Document, document=document, version=version)
    return version


# ── AI insight review (REPO-SP9 / SRS FRREP-AR007-AR008) ────────────────────


def _extract_year(value) -> int | None:
    """Best-effort year extraction from a free-form ``publication_date`` string."""
    if not value:
        return None
    match = re.search(r"(19|20)\d{2}", str(value))
    if not match:
        return None
    return int(match.group(0))


@transaction.atomic
def decide_ai_insight(insight, *, decision: str, actor, request=None):
    """Human review of an AI classification/metadata suggestion.

    ``decision="accept"`` marks the insight accepted and *applies* it:
    classification moves the document to the suggested collection; metadata
    fills in currently-blank document fields only (never overwrites a
    human-entered value). ``decision="dismiss"`` records the rejection with
    no other mutation — the manager can still edit metadata by hand via the
    existing edit flow. Notifies the document's submitter on accept.
    """
    from apps.repository.analytics.models import AIDocumentInsight

    if decision not in {"accept", "dismiss"}:
        raise ValidationError("decision must be 'accept' or 'dismiss'.")
    if insight.accepted_by_user is not None:
        return insight  # already decided — idempotent no-op

    document = insight.document
    changes: dict[str, Any] = {"kind": insight.kind, "decision": decision}

    if decision == "dismiss":
        insight.accepted_by_user = False
        insight.save(update_fields=["accepted_by_user"])
        audit_repository(
            actor=actor,
            action=action_update(),
            target_model="AIDocumentInsight",
            object_id=insight.pk,
            object_repr=str(document),
            changes=changes,
            request=request,
        )
        return insight

    insight.accepted_by_user = True
    insight.save(update_fields=["accepted_by_user"])

    if insight.kind == AIDocumentInsight.Kind.CLASSIFICATION:
        slug = (insight.payload or {}).get("suggested_collection_slug", "")
        collection = Collection.objects.filter(slug=slug).first() if slug else None
        if collection and collection.pk != document.collection_id:
            changes["collection"] = {"old": document.collection.slug, "new": collection.slug}
            document.collection = collection
            document.save(update_fields=["collection", "updated_at"])
    elif insight.kind == AIDocumentInsight.Kind.METADATA:
        payload = insight.payload or {}
        update_fields: list[str] = []

        if not document.year:
            year = _extract_year(payload.get("publication_date"))
            if year:
                document.year = year
                update_fields.append("year")
                changes["year"] = year

        if not document.authors.exists():
            names = [n for n in (payload.get("authors") or []) if isinstance(n, str) and n.strip()]
            if names:
                new_authors = []
                for author_name in names:
                    first, last = _split_author_name(author_name)
                    author = register_author(first_name=first, last_name=last)
                    if author is not None:
                        new_authors.append(author)
                if new_authors:
                    set_document_authors(document, new_authors, actor=actor, request=request)
                    changes["authors"] = [a.full_name for a in new_authors]

        if update_fields:
            update_fields.append("updated_at")
            document.save(update_fields=update_fields)

    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="AIDocumentInsight",
        object_id=insight.pk,
        object_repr=str(document),
        changes=changes,
        request=request,
    )
    if document.submitted_by_id and document.submitted_by_id != getattr(actor, "pk", None):
        actor_name = actor.get_full_name() or actor.email
        kind_label = insight.get_kind_display().lower()
        send_notification(
            document.submitted_by,
            f'{actor_name} accepted an AI {kind_label} suggestion on "{document.title[:80]}".',
            verb=Notification.Verb.REPO_AI_INSIGHT_ACCEPTED,
        )
    return insight


# ── metadata edit with optimistic-lock (REPO-SP7 FRREP-COL008) ──────────────


class EditConflictError(ValidationError):
    """Raised when a metadata edit is submitted against a stale document state."""

    def __init__(self, current_token: str):
        super().__init__(
            "This document was modified by someone else while you were editing it."
        )
        self.current_token = current_token


def document_edit_token(document: Document) -> str:
    """A coarse-grained ETag used to detect concurrent metadata edits."""
    if not document.updated_at:
        return "0"
    # Microsecond precision is sufficient — auto_now bumps on every save().
    return document.updated_at.isoformat()


def _split_author_name(name_part: str) -> tuple[str, str]:
    """Split a display name into (first, last).

    ``Last, First`` (comma form) and ``First Last`` (space form) are both
    accepted, matching the legacy free-text author block and the registry-first
    picker's new-author form.
    """
    name_part = (name_part or "").strip()
    if "," in name_part:
        last, first = [p.strip() for p in name_part.split(",", 1)]
        return first, last
    tokens = name_part.split()
    first = tokens[0] if tokens else ""
    last = " ".join(tokens[1:]) if len(tokens) > 1 else ""
    return first, last


def parse_authors_text(text: str) -> list[Author]:
    """Parse the wizard's free-text author block into resolved ``Author`` rows.

    Format: one author per line — ``Last, First | ORCID | email``. Pipes are
    optional; a blank ORCID/email is allowed. Each line is run through
    :func:`register_author` so duplicates dedupe by ORCID, then by email,
    then by case-insensitive name, then by ``AuthorAlias`` history
    (FRREP-MCL004 name-change linking).

    Retained as the no-JS / edit-form fallback; the submit wizard now posts a
    structured selection through :func:`resolve_authors_payload`.
    """
    result: list[Author] = []
    for raw in (text or "").splitlines():
        line = raw.strip()
        if not line:
            continue
        parts = [p.strip() for p in line.split("|")]
        first, last = _split_author_name(parts[0])
        orcid = parts[1] if len(parts) > 1 and parts[1] else None
        email = parts[2] if len(parts) > 2 else ""
        result.append(
            register_author(
                first_name=first,
                last_name=last,
                orcid=orcid,
                email=email,
            )
        )
    return result


def _iter_author_payload(raw: str) -> list[dict]:
    """Parse the wizard's ``authors_json`` hidden field into a list of dicts.

    Tolerant of junk — returns ``[]`` for anything that isn't a JSON array of
    objects. Does **not** touch the database, so it is safe to call from form
    ``clean()`` (validation must not create Author rows).
    """
    import json

    try:
        items = json.loads(raw or "[]")
    except (ValueError, TypeError):
        return []
    if not isinstance(items, list):
        return []
    return [item for item in items if isinstance(item, dict)]


def count_payload_authors(raw: str) -> int:
    """How many resolvable authors the structured payload carries.

    An entry counts when it names an existing registry author (``id``) or
    supplies a non-empty new-author ``name``. Read-only — used by the form's
    ≥1-author check before ingest.
    """
    count = 0
    for item in _iter_author_payload(raw):
        if item.get("id") or (item.get("name") or "").strip():
            count += 1
    return count


def resolve_authors_payload(raw: str) -> list[Author]:
    """Resolve the registry-first picker's selection into ordered ``Author`` rows.

    ``raw`` is a JSON array; each item is either an existing registry author
    ``{"id": <pk>}`` or a brand-new author
    ``{"name", "orcid", "email", "affiliation"}``. Existing ids attach that exact
    Author with no re-dedup; new entries run through :func:`register_author`, so
    ORCID/email/name dedupe and the ORCID verification hook (FRREP-MCL002) still
    apply. Affiliation text is captured on new authors (FRREP-MCL003). Order is
    preserved and duplicates are collapsed.
    """
    result: list[Author] = []
    seen: set[int] = set()
    for item in _iter_author_payload(raw):
        pk = item.get("id")
        if pk:
            author = Author.objects.filter(pk=pk).first()
        else:
            name = (item.get("name") or "").strip()
            if not name:
                continue
            first, last = _split_author_name(name)
            author = register_author(
                first_name=first,
                last_name=last,
                orcid=(item.get("orcid") or "").strip() or None,
                email=(item.get("email") or "").strip(),
                affiliation_text=(item.get("affiliation") or "").strip(),
            )
        if author is None or author.pk in seen:
            continue
        seen.add(author.pk)
        result.append(author)
    return result


def render_authors_text(document: Document) -> str:
    """Round-trip a document's authors back into the wizard's text format so
    the edit form can pre-populate the textarea."""
    lines: list[str] = []
    for da in (
        document.documentauthor_set
        .select_related("author")
        .order_by("order")
    ):
        a = da.author
        name = f"{a.last_name}, {a.first_name}".strip(", ")
        parts = [name]
        if a.orcid:
            parts.append(a.orcid)
            if a.email:
                parts.append(a.email)
        elif a.email:
            parts.extend(["", a.email])
        lines.append(" | ".join(parts))
    return "\n".join(lines)


@transaction.atomic
def set_document_authors(
    document: Document,
    authors: list[Author],
    *,
    actor=None,
    request=None,
) -> list[DocumentAuthor]:
    """Replace the document's author roster (FRREP-MCL004 / FRREP-MCL011).

    Strategy:

    1. Build the desired set of ``Author`` ids in the supplied order.
    2. Delete ``DocumentAuthor`` rows whose ``author_id`` is not in the new
       set.
    3. For each (index, author) pair, ``update_or_create`` the row with the
       new ``order`` and the right ``role`` (PRIMARY at index 0, CO_AUTHOR
       elsewhere).

    Idempotent: re-running with the same list is a no-op aside from the
    audit row. Empty list is rejected — every Document needs at least one
    author per FRREP-DCI010.
    """
    if not authors:
        raise ValidationError("A document needs at least one author.")

    desired_ids = [a.pk for a in authors]
    seen: set[int] = set()
    deduped: list[Author] = []
    for a in authors:
        if a.pk in seen:
            continue
        seen.add(a.pk)
        deduped.append(a)
    desired_ids = [a.pk for a in deduped]

    previous = list(
        document.documentauthor_set
        .select_related("author")
        .order_by("order")
        .values_list("author_id", flat=True)
    )

    document.documentauthor_set.exclude(author_id__in=desired_ids).delete()

    rows: list[DocumentAuthor] = []
    for idx, author in enumerate(deduped):
        role = (
            DocumentAuthor.AuthorRole.PRIMARY
            if idx == 0
            else DocumentAuthor.AuthorRole.CO_AUTHOR
        )
        row, _ = DocumentAuthor.objects.update_or_create(
            document=document,
            author=author,
            defaults={"order": idx, "role": role},
        )
        rows.append(row)

    if previous != desired_ids:
        audit_repository(
            actor=actor,
            action=action_update(),
            target_model="Document",
            object_id=document.pk,
            object_repr=document.public_document_id,
            changes={
                "authors_previous": previous,
                "authors_new": desired_ids,
            },
            request=request,
        )
        # FRREP-MCL011 — author names are part of the FTS vector; reindex when
        # the roster actually changes so author-name searches stay accurate.
        from apps.repository.search.indexes import update_search_vector

        transaction.on_commit(lambda doc_id=document.pk: update_search_vector(doc_id))

    return rows


@transaction.atomic
def update_document_metadata(
    document: Document,
    *,
    fields: dict[str, Any],
    expected_token: str,
    actor,
    request=None,
) -> Document:
    """Update Document metadata fields, refusing the change if the token is stale.

    ``document`` is typically the instance backing a bound ``ModelForm`` — and
    Django's ``_post_clean()`` (run inside ``form.is_valid()``, *before* this
    function is ever called) already calls ``construct_instance()``, which
    ``setattr``s every one of the form's cleaned field values onto the
    instance regardless of whether we ever call ``form.save()``. That means
    ``getattr(document, field_name)`` can no longer be trusted as the "old"
    value — it may already equal ``new_value``, which previously made every
    edit look like a no-op and silently skipped the ``.save()`` call
    entirely. The true old values are re-read from the database instead.
    """
    if not fields:
        return document
    current_token = document_edit_token(document)
    if expected_token and expected_token != current_token:
        raise EditConflictError(current_token)

    old_values = dict(
        Document.objects.filter(pk=document.pk).values(*fields.keys()).first() or {}
    )

    changes: dict[str, dict[str, Any]] = {}
    for field_name, new_value in fields.items():
        old_value = old_values.get(field_name)
        if old_value != new_value:
            changes[field_name] = {"old": old_value, "new": new_value}
        setattr(document, field_name, new_value)
    if not changes:
        return document
    # FRREP-COL008 — record the winner synchronously; `notify_edit_conflict`
    # needs this the instant a *different* editor's save is rejected, and the
    # audit trail (written via Celery after commit) isn't available that fast.
    update_fields = list(fields.keys()) + ["updated_at"]
    if actor is not None and getattr(actor, "pk", None):
        document.last_edited_by = actor
        update_fields.append("last_edited_by")
    document.save(update_fields=update_fields)
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes=changes,
        request=request,
    )
    # FRREP-MCL011 — refresh the Postgres FTS vector so an edited title/abstract/
    # metadata value is reflected in search immediately. The publish-time
    # receiver only fires on the *transition* into PUBLISHED, so post-publish
    # edits would otherwise leave search results stale indefinitely.
    from apps.repository.search.indexes import update_search_vector

    transaction.on_commit(lambda doc_id=document.pk: update_search_vector(doc_id))
    return document


def notify_edit_conflict(document: Document, *, rejected_editor) -> None:
    """FRREP-COL008 — a stale-token save was rejected by `update_document_metadata`
    and the rejected edit's changes are simply discarded (no two-version
    storage — that's out of scope). At minimum, tell both sides a conflict
    happened: the editor whose save was rejected, and whoever's edit actually
    won the race (`Document.last_edited_by`, updated synchronously by every
    successful `update_document_metadata` call — the audit trail is written
    asynchronously via Celery so it can't be read back this quickly).
    """
    User = get_user_model()
    winner_id = document.last_edited_by_id
    recipient_ids = {uid for uid in (getattr(rejected_editor, "pk", None), winner_id) if uid}
    if not recipient_ids:
        return
    message = (
        f'A concurrent edit conflict occurred on "{document.title[:80]}" — two '
        "people edited it at nearly the same time. Please reload the document "
        "and re-apply any changes that didn't save."
    )
    for user in User.objects.filter(pk__in=recipient_ids):
        send_notification(user, message, verb=Notification.Verb.REPO_EDIT_CONFLICT)


# ── keyword vocabulary (REPO-SP2 FRREP-MCL007) ──────────────────────────────


def _slugify_label(label: str) -> str:
    from django.utils.text import slugify

    return slugify(label)[:96] or "tag"


@transaction.atomic
def suggest_or_get_keyword(label: str, *, suggested_by) -> Keyword:
    """Reuse an existing approved/pending keyword or create a new pending one."""
    label = label.strip()
    if not label:
        raise ValidationError("Keyword label is required.")
    slug = _slugify_label(label)
    keyword, created = Keyword.objects.get_or_create(
        slug=slug,
        defaults={
            "label": label,
            "status": Keyword.Status.PENDING,
            "suggested_by": suggested_by,
        },
    )
    return keyword


@transaction.atomic
def review_keyword(
    keyword: Keyword,
    *,
    approve: bool,
    actor,
    request=None,
) -> Keyword:
    keyword.status = (
        Keyword.Status.APPROVED if approve else Keyword.Status.REJECTED
    )
    keyword.reviewed_by = actor
    keyword.reviewed_at = timezone.now()
    keyword.save(update_fields=["status", "reviewed_by", "reviewed_at", "updated_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Keyword",
        object_id=keyword.pk,
        object_repr=keyword.label,
        changes={"status": keyword.status},
        request=request,
    )
    if not approve and keyword.suggested_by_id:
        send_notification(
            keyword.suggested_by,
            f'Tag suggestion "{keyword.label}" was rejected by repository staff.',
            verb=Notification.Verb.REPO_TAG_SUGGESTION_PENDING,
        )
    return keyword


def apply_keywords(document: Document, labels: list[str], *, suggested_by) -> list[Keyword]:
    """Attach approved keywords (or queue suggestions for new ones)."""
    out: list[Keyword] = []
    for raw in labels:
        label = (raw or "").strip()
        if not label:
            continue
        keyword = suggest_or_get_keyword(label, suggested_by=suggested_by)
        if keyword.is_approved:
            document.keywords.add(keyword)
        out.append(keyword)
    return out


# ── auto-classification rule engine (FRREP-MCL006) ──────────────────────────


def evaluate_auto_classification(
    *, keywords: list[str] | None = None, title: str = "", abstract: str = ""
) -> Collection | None:
    """Suggest a target collection from the configured classification rules.

    Rules live on :class:`MetadataSchema.auto_classification_rules` as a list of
    ``{"if_keyword": str, "then_collection_slug": str}``. Every schema's rules
    are treated as one global ruleset. A rule fires when its ``if_keyword``
    matches one of the document's keywords (case-insensitive, exact) or appears
    as a substring of the title/abstract. The first firing rule whose target
    collection exists wins. Returns ``None`` when nothing matches.

    Advisory only — the caller surfaces the suggestion and the user accepts or
    overrides it (overrides are audited via :func:`log_classification_override`).
    """
    from apps.repository.documents.models import MetadataSchema

    keyword_set = {(k or "").strip().lower() for k in (keywords or []) if (k or "").strip()}
    haystack = f"{title or ''} {abstract or ''}".lower()

    for schema in MetadataSchema.objects.exclude(auto_classification_rules=[]):
        for rule in schema.auto_classification_rules or []:
            if not isinstance(rule, dict):
                continue
            needle = (rule.get("if_keyword") or "").strip().lower()
            slug = (rule.get("then_collection_slug") or "").strip()
            if not needle or not slug:
                continue
            if needle in keyword_set or (needle in haystack):
                collection = Collection.objects.filter(slug=slug).first()
                if collection is not None:
                    return collection
    return None


def log_classification_override(
    document: Document,
    *,
    suggested_slug: str,
    chosen_slug: str,
    actor=None,
    request=None,
) -> None:
    """Record that the uploader kept their collection over the system's
    suggestion (SRS A3 — overrides are auditable)."""
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={
            "auto_classification_suggested": suggested_slug,
            "auto_classification_chosen": chosen_slug,
            "auto_classification_overridden": True,
        },
        request=request,
    )


# ── grant closure flag (REPO-SP5 / SP8) ─────────────────────────────────────


@transaction.atomic
def apply_grant_closure_flag(
    grant_reference: str, *, actor=None, request=None
) -> int:
    """Flag every Document linked to a closed RIMS grant (PRD REPO-SP5/SP8).

    Called by :class:`apps.repository.integration.rims_api.GrantClosureNotifyView`
    when RIMS's grant close-out flow fires. FRREP-VL010 — audited like every
    other bulk lifecycle mutation in this module.
    """
    if not grant_reference:
        return 0
    doc_ids = list(
        Document.objects.filter(grant_reference=grant_reference).values_list(
            "pk", flat=True
        )
    )
    if not doc_ids:
        return 0
    updated = Document.objects.filter(pk__in=doc_ids).update(
        grant_closed_flag=True,
        updated_at=timezone.now(),
    )
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=None,
        object_repr=grant_reference,
        changes={
            "grant_closed_flag": True,
            "grant_reference": grant_reference,
            "document_ids": doc_ids,
            "count": updated,
        },
        request=request,
    )
    return updated


@transaction.atomic
def archive_document(
    document: Document, *, reason: str = "", actor=None, request=None
) -> Document:
    """FRREP-VL006 — retention-policy "archive" action.

    Sets ``is_archived`` (hides the document from Browse/search) without
    touching its FSM status — an archived PUBLISHED document is still
    PUBLISHED, just no longer surfaced by default. Idempotent.
    """
    if document.is_archived:
        return document
    document.is_archived = True
    document.archived_at = timezone.now()
    document.save(update_fields=["is_archived", "archived_at", "updated_at"])
    audit_repository(
        actor=actor,
        action=action_update(),
        target_model="Document",
        object_id=document.pk,
        object_repr=document.public_document_id,
        changes={"is_archived": True, "reason": reason},
        request=request,
    )
    return document


# ── outbox helper used by receivers and integration sub-module ──────────────


def enqueue_outbox(*, target: str, event_type: str, payload: dict) -> IntegrationOutbox:
    return IntegrationOutbox.objects.create(
        target=target, event_type=event_type, payload=payload
    )


# ---------------------------------------------------------------------------
# Resumable / chunked uploads (FRREP-DCI004)
# ---------------------------------------------------------------------------


RESUMABLE_DIR = "resumable"
RESUMABLE_CHUNK_SIZE_BYTES = 5 * 1024 * 1024  # 5 MB
RESUMABLE_SESSION_TTL_HOURS = 24


class ResumableUploadError(Exception):
    """Base class for resumable upload service errors."""


class ResumableUploadNotFound(ResumableUploadError):
    """The token does not exist or has expired."""


class ResumableUploadHashMismatch(ResumableUploadError):
    """The assembled file's SHA-256 did not match the client-supplied digest."""


class ResumableUploadValidationFailed(ResumableUploadError):
    """The assembled file failed magic-bytes / size / category validation."""

    def __init__(self, message: str):
        super().__init__(message)
        self.message = message


def _resumable_chunk_dir(token: str) -> Path:
    return Path(settings.MEDIA_ROOT) / RESUMABLE_DIR / token


def _staged_dir(token: str) -> Path:
    # Match the constants defined in apps.repository.documents.views.
    return Path(settings.MEDIA_ROOT) / "staged_uploads" / token


def _effective_module_cap_mb(extension: str) -> float:
    config = FILE_TYPE_REGISTRY.get(extension.lower())
    per_type = config.max_size_mb if config else DEFAULT_MAX_SIZE_MB
    module = MODULE_SIZE_LIMITS.get("repository", DEFAULT_MAX_SIZE_MB)
    return float(min(per_type, module))


def start_resumable_upload(
    *,
    user,
    filename: str,
    total_bytes: int,
    chunk_size: int = RESUMABLE_CHUNK_SIZE_BYTES,
) -> ResumableUpload:
    """Open a new resumable session.

    Pre-flight: extension is in the registry; total size is within the
    effective module cap (tighter of per-extension and module limits).
    """
    if total_bytes <= 0:
        raise ValidationError("total_bytes must be positive.")
    if chunk_size <= 0:
        raise ValidationError("chunk_size must be positive.")

    safe_name = get_valid_filename(filename) or "upload"
    _, ext = os.path.splitext(safe_name.lower())
    if filename.lower().endswith(".tar.gz"):
        ext = ".tar.gz"
    if ext not in FILE_TYPE_REGISTRY:
        raise ValidationError(
            f"File type '{ext}' is not supported."
        )
    # Mirror the direct-upload category gate (#2, FRREP-DCI005) so chunked
    # uploads can't smuggle in audio/video by skipping _validate_repository_upload.
    category = FILE_TYPE_REGISTRY[ext].category
    if category not in REPOSITORY_UPLOAD_CATEGORIES:
        allowed = ", ".join(REPOSITORY_UPLOAD_CATEGORIES)
        raise ValidationError(
            f"File type '{ext}' is not allowed in the repository. "
            f"Accepted formats are documents, data, images and archives ({allowed})."
        )

    cap_mb = _effective_module_cap_mb(ext)
    if total_bytes > cap_mb * 1024 * 1024:
        raise ValidationError(
            f"File size exceeds the {cap_mb:.0f} MB limit for the repository module."
        )

    expected_chunks = (total_bytes + chunk_size - 1) // chunk_size
    token = uuid.uuid4().hex
    upload = ResumableUpload.objects.create(
        token=token,
        user=user,
        filename=safe_name,
        total_bytes=total_bytes,
        chunk_size=chunk_size,
        expected_chunks=expected_chunks,
        expires_at=timezone.now() + timedelta(hours=RESUMABLE_SESSION_TTL_HOURS),
    )
    _resumable_chunk_dir(token).mkdir(parents=True, exist_ok=True)
    return upload


def _resolve_active_upload(token: str, user) -> ResumableUpload:
    try:
        upload = ResumableUpload.objects.get(token=token, user=user)
    except ResumableUpload.DoesNotExist as exc:
        raise ResumableUploadNotFound("Unknown resumable token.") from exc
    if upload.status == ResumableUpload.Status.COMPLETED:
        raise ResumableUploadNotFound("Upload session already completed.")
    if upload.status == ResumableUpload.Status.ABANDONED:
        raise ResumableUploadNotFound("Upload session was abandoned.")
    if upload.expires_at <= timezone.now():
        raise ResumableUploadNotFound("Upload session has expired.")
    return upload


@transaction.atomic
def append_resumable_chunk(*, token: str, user, chunk_index: int, body: bytes) -> ResumableUpload:
    """Persist one chunk. Idempotent — re-sending the same index is a no-op."""
    if chunk_index < 0:
        raise ValidationError("chunk_index must be non-negative.")
    try:
        upload = ResumableUpload.objects.select_for_update().get(token=token, user=user)
    except ResumableUpload.DoesNotExist as exc:
        raise ResumableUploadNotFound("Unknown resumable token.") from exc
    if upload.status != ResumableUpload.Status.IN_PROGRESS:
        raise ResumableUploadNotFound("Upload session is not active.")
    if upload.expires_at <= timezone.now():
        raise ResumableUploadNotFound("Upload session has expired.")
    if chunk_index >= upload.expected_chunks:
        raise ValidationError("chunk_index exceeds expected_chunks.")

    chunk_dir = _resumable_chunk_dir(token)
    chunk_dir.mkdir(parents=True, exist_ok=True)
    chunk_path = chunk_dir / f"chunk_{chunk_index:06d}"

    received = list(upload.received_chunk_indexes or [])
    if chunk_index in received:
        # Idempotent: the chunk is already on disk; return current state.
        return upload

    with chunk_path.open("wb") as fh:
        fh.write(body)

    received.append(chunk_index)
    upload.received_chunk_indexes = sorted(received)
    upload.chunks_received = len(received)
    upload.save(update_fields=["received_chunk_indexes", "chunks_received", "updated_at"])
    return upload


def finalize_resumable_upload(
    *,
    token: str,
    user,
    expected_sha256: str | None = None,
) -> tuple[ResumableUpload, dict]:
    """Concatenate chunks, validate the assembled file, register it as a
    staged-upload entry, and return the staged-upload metadata.

    Returns (upload, staged_entry) — the caller (view) registers
    ``staged_entry`` in the session under the same token so the existing
    submission wizard consumes it.

    Not wrapped in a single ``@transaction.atomic``: validation can raise
    after disk IO and we want the ABANDONED state to persist. Each status
    update runs in its own atomic block.
    """
    try:
        upload = ResumableUpload.objects.get(token=token, user=user)
    except ResumableUpload.DoesNotExist as exc:
        raise ResumableUploadNotFound("Unknown resumable token.") from exc
    if upload.status == ResumableUpload.Status.COMPLETED:
        raise ResumableUploadNotFound("Upload session already completed.")
    if upload.status == ResumableUpload.Status.ABANDONED:
        raise ResumableUploadNotFound("Upload session was abandoned.")
    if upload.chunks_received != upload.expected_chunks:
        raise ValidationError(
            f"Cannot finalize — {upload.chunks_received} of "
            f"{upload.expected_chunks} chunks received."
        )

    chunk_dir = _resumable_chunk_dir(token)
    staged_dir = _staged_dir(token)
    staged_dir.mkdir(parents=True, exist_ok=True)
    assembled_path = staged_dir / upload.filename

    def _abort_and_raise(exc):
        with transaction.atomic():
            _abandon_resumable(
                upload, cleanup_chunks=True, cleanup_assembled=assembled_path,
            )
        raise exc

    h = hashlib.sha256()
    with assembled_path.open("wb") as out_fh:
        for index in range(upload.expected_chunks):
            chunk_path = chunk_dir / f"chunk_{index:06d}"
            if not chunk_path.exists():
                _abort_and_raise(ValidationError(f"Chunk {index} missing from disk."))
            with chunk_path.open("rb") as in_fh:
                for block in iter(lambda fh=in_fh: fh.read(65536), b""):
                    out_fh.write(block)
                    h.update(block)

    sha256_hex = h.hexdigest()
    if expected_sha256 and expected_sha256.lower() != sha256_hex:
        _abort_and_raise(
            ResumableUploadHashMismatch(
                f"Hash mismatch: client {expected_sha256} vs server {sha256_hex}.",
            ),
        )

    actual_size = assembled_path.stat().st_size
    if actual_size != upload.total_bytes:
        _abort_and_raise(
            ValidationError(
                f"Assembled size {actual_size} does not match declared {upload.total_bytes}.",
            ),
        )

    with assembled_path.open("rb") as fh:
        django_file = DjangoFile(fh, name=upload.filename)
        django_file.size = actual_size
        try:
            _validate_repository_upload(django_file, actor=user)
        except ValidationError as exc:
            msg = exc.message if hasattr(exc, "message") else str(exc)
            _abort_and_raise(ResumableUploadValidationFailed(msg))

    with transaction.atomic():
        upload.sha256_hex = sha256_hex
        upload.status = ResumableUpload.Status.COMPLETED
        upload.save(update_fields=["sha256_hex", "status", "updated_at"])

    _safe_rmtree(chunk_dir)

    rel_path = Path("staged_uploads") / token / upload.filename
    staged_entry = {
        "rel_path": str(rel_path),
        "name": upload.filename,
        "size": actual_size,
        "content_type": "",
    }
    return upload, staged_entry


@transaction.atomic
def abort_resumable_upload(*, token: str, user) -> ResumableUpload | None:
    """Mark a session ABANDONED and remove its chunk directory.

    Returns ``None`` when the token is unknown — callers treat that as a
    benign no-op.
    """
    upload = ResumableUpload.objects.filter(token=token, user=user).first()
    if upload is None:
        return None
    _abandon_resumable(upload, cleanup_chunks=True)
    return upload


def _abandon_resumable(
    upload: ResumableUpload,
    *,
    cleanup_chunks: bool = True,
    cleanup_assembled: Path | None = None,
) -> None:
    if upload.status != ResumableUpload.Status.ABANDONED:
        upload.status = ResumableUpload.Status.ABANDONED
        upload.save(update_fields=["status", "updated_at"])
    if cleanup_chunks:
        _safe_rmtree(_resumable_chunk_dir(upload.token))
    if cleanup_assembled is not None:
        try:
            if cleanup_assembled.exists():
                cleanup_assembled.unlink()
            parent = cleanup_assembled.parent
            if parent.exists() and not any(parent.iterdir()):
                parent.rmdir()
        except OSError:
            pass


def _safe_rmtree(path: Path) -> None:
    try:
        if path.exists():
            shutil.rmtree(path)
    except OSError:
        logger.warning("resumable: failed to remove %s", path, exc_info=True)


def cleanup_expired_resumable_uploads() -> int:
    """Remove abandoned and expired in-progress sessions and their chunk dirs.

    Returns the number of rows touched. Designed to be called from a Celery
    beat task.
    """
    now = timezone.now()
    stale = ResumableUpload.objects.filter(
        models.Q(status=ResumableUpload.Status.ABANDONED)
        | models.Q(
            status=ResumableUpload.Status.IN_PROGRESS,
            expires_at__lt=now,
        )
    )
    count = 0
    for upload in stale:
        _safe_rmtree(_resumable_chunk_dir(upload.token))
        if upload.status == ResumableUpload.Status.IN_PROGRESS:
            upload.status = ResumableUpload.Status.ABANDONED
            upload.save(update_fields=["status", "updated_at"])
        count += 1
    return count