Fix: exclude version documents from bulk edit "select all matching"

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
   "djangorestframework~=3.16",
   "drf-spectacular~=0.30",
   "drf-spectacular-sidecar~=2026.7.1",
+  "drf-writable-nested~=0.7.1",
   "filelock~=3.32.0",
   "flower~=2.0.1",
   "gotenberg-client~=0.14.0",

src-ui/src/app/components/app-frame/app-frame.component.scss

@@ -417,7 +417,7 @@ main {
 
 :host ::ng-deep .navbar-official-logo {
   .leaf {
-    fill: color-mix(in srgb, var(--pngx-primary-text-contrast) 70%, var(--bs-primary)) !important;
+    fill: color-mix(in srgb, var(--pngx-primary-text-contrast) 85%, var(--bs-primary)) !important;
   }
 
   .text {

src/documents/bulk_edit.py

@@ -27,7 +27,7 @@ from documents.models import DocumentType
 from documents.models import PaperlessTask
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import set_permissions_for_objects
+from documents.permissions import set_permissions_for_object
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
@@ -305,49 +305,33 @@ def modify_custom_fields(
         else [(field, None) for field in add_custom_fields]
     )
 
-    custom_fields_by_id: dict[int, CustomField] = {
-        cf.id: cf
-        for cf in CustomField.objects.filter(
-            id__in=[int(field) for field, _ in add_custom_fields],
-        )
-    }
-    # Deferred, not `.only()`: these objects get cached onto the FK
-    # descriptor of newly-created CustomFieldInstance rows below, and
-    # downstream post_save receivers (e.g. the filename-generation signal)
-    # touch other Document fields -- `.only("pk")` would just turn that into
-    # a deferred-field reload per document, trading one N+1 for another.
-    # `content` is the one field guaranteed to be both large (full OCR text)
-    # and unused by anything this function or its receivers touch.
-    docs_by_id: dict[int, Document] = {
-        doc.id: doc
-        for doc in Document.objects.filter(id__in=affected_docs).defer("content")
-    }
+    custom_fields = CustomField.objects.filter(
+        id__in=[int(field) for field, _ in add_custom_fields],
+    ).distinct()
     for field_id, value in add_custom_fields:
-        custom_field = custom_fields_by_id[field_id]
-        value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
-            custom_field.data_type
-        ]
         for doc_id in affected_docs:
-            defaults = {value_field: value}
-            if (
-                custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
-                and value
-                and doc_id in value
-            ):
-                # Prevent self-linking
-                continue
-            # Pass the already-resolved objects, not bare ids: this caches
-            # them on the FK descriptor of any newly-created instance, so a
-            # later `.field`/`.document` access (e.g. auditlog's post_save
-            # receiver calling `str(instance)`, which touches `.field.name`)
-            # doesn't trigger its own per-instance re-fetch.
+            defaults = {}
+            custom_field = custom_fields.get(id=field_id)
+            if custom_field:
+                value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
+                    custom_field.data_type
+                ]
+                defaults[value_field] = value
+                if (
+                    custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
+                    and value
+                    and doc_id in value
+                ):
+                    # Prevent self-linking
+                    continue
             CustomFieldInstance.objects.update_or_create(
-                document=docs_by_id[doc_id],
-                field=custom_field,
+                document_id=doc_id,
+                field_id=field_id,
                 defaults=defaults,
             )
             if custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
-                reflect_doclinks(docs_by_id[doc_id], custom_field, value)
+                doc = Document.objects.get(id=doc_id)
+                reflect_doclinks(doc, custom_field, value)
 
     # For doc link fields that are being removed, remove symmetrical links
     for doclink_being_removed_instance in CustomFieldInstance.objects.filter(
@@ -355,10 +339,12 @@ def modify_custom_fields(
         field__id__in=remove_custom_fields,
         field__data_type=CustomField.FieldDataType.DOCUMENTLINK,
         value_document_ids__isnull=False,
-    ).select_related("field"):
+    ):
         for target_doc_id in doclink_being_removed_instance.value:
             remove_doclink(
-                document=docs_by_id[doclink_being_removed_instance.document_id],
+                document=Document.objects.get(
+                    id=doclink_being_removed_instance.document.id,
+                ),
                 field=doclink_being_removed_instance.field,
                 target_doc_id=target_doc_id,
             )
@@ -444,13 +430,10 @@ def set_permissions(
     else:
         qs.update(owner=owner)
 
+    for doc in qs:
+        set_permissions_for_object(permissions=set_permissions, object=doc, merge=merge)
+
     affected_docs = list(qs.values_list("pk", flat=True))
-    set_permissions_for_objects(
-        permissions=set_permissions,
-        model=Document,
-        pks=affected_docs,
-        merge=merge,
-    )
 
     bulk_update_documents.apply_async(
         kwargs={"document_ids": affected_docs},

src/documents/data_models.py

@@ -129,7 +129,7 @@ class DocumentMetadataOverrides:
         )
         overrides.custom_fields = {
             custom_field.field.id: custom_field.value
-            for custom_field in doc.custom_fields.select_related("field").all()
+            for custom_field in doc.custom_fields.all()
         }
 
         groups_with_perms = get_groups_with_perms(

src/documents/permissions.py

@@ -173,182 +173,6 @@ def set_permissions_for_object(
                         )
 
 
-def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]:
-    """
-    Resolves `codenames` to Permission rows, raising like the single-object
-    assign_perm() this bulk path replaces does (via a `.get()` internally)
-    if any codename doesn't exist -- e.g. a client-supplied action name that
-    was never validated (BulkEditObjectsSerializer._validate_permissions
-    calls validate_set_permissions() only for its side-effecting id checks
-    and discards the filtered dict it returns, so an unrecognized action key
-    reaches this function as-is). A plain `.filter()` with no existence
-    check would otherwise silently build zero rows and no-op instead of
-    reporting the bad input.
-    """
-    permission_objs = list(
-        Permission.objects.filter(content_type=ctype, codename__in=codenames),
-    )
-    missing = codenames - {p.codename for p in permission_objs}
-    if missing:
-        raise Permission.DoesNotExist(
-            f"Permission matching query does not exist for codename(s): "
-            f"{', '.join(sorted(missing))}",
-        )
-    return permission_objs
-
-
-# Target number of permission rows to build in Python before handing them to
-# bulk_create -- keeps peak memory bounded for a large "apply to all" call,
-# independent of bulk_create's own batch_size (which only caps the size of
-# each INSERT statement, not how many row objects exist in memory at once).
-_PERMISSION_ROW_CHUNK_SIZE = 5000
-
-
-def _apply_bulk_permission_entry(
-    *,
-    perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
-    identity_model: type[User] | type[Group],
-    identity_field: str,
-    ids: list[int],
-    codename: str,
-    permission_objs: list[Permission],
-    ctype: ContentType,
-    object_pks: list[str],
-    merge: bool,
-) -> None:
-    # Only the ids are needed to build permission rows (via `<field>_id=`),
-    # so avoid fetching full User/Group rows for identities that may not
-    # even end up being granted anything new.
-    add_ids = set(
-        identity_model.objects.filter(id__in=ids).values_list("id", flat=True),
-    )
-
-    if not merge:
-        existing_ids = set(
-            perm_model.objects.filter(
-                content_type=ctype,
-                object_pk__in=object_pks,
-                permission__codename=codename,
-            ).values_list(f"{identity_field}_id", flat=True),
-        )
-        remove_ids = existing_ids - add_ids
-        if remove_ids:
-            perm_model.objects.filter(
-                content_type=ctype,
-                object_pk__in=object_pks,
-                permission__codename=codename,
-                **{f"{identity_field}_id__in": remove_ids},
-            ).delete()
-
-    if not add_ids:
-        return
-
-    rows_per_pk = len(permission_objs) * len(add_ids)
-    pks_per_chunk = max(1, _PERMISSION_ROW_CHUNK_SIZE // rows_per_pk)
-    for start in range(0, len(object_pks), pks_per_chunk):
-        pk_chunk = object_pks[start : start + pks_per_chunk]
-        rows = [
-            perm_model(
-                content_type=ctype,
-                object_pk=pk,
-                permission=permission_obj,
-                **{f"{identity_field}_id": identity_id},
-            )
-            for permission_obj in permission_objs
-            for pk in pk_chunk
-            for identity_id in add_ids
-        ]
-        # ignore_conflicts skips only rows that already exist as an exact
-        # (identity, permission, object) match -- the same de-dup the
-        # underlying (user|group, permission, object_pk) unique constraint
-        # already enforces for the single-object assign_perm() this
-        # replaces, so it doesn't change what counts as "already granted".
-        # batch_size caps how many rows go into a single INSERT so a huge
-        # chunk doesn't build one enormous statement.
-        perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
-
-
-def set_permissions_for_objects(
-    permissions: dict,
-    model: type[Model],
-    pks: QuerySet | list,
-    *,
-    merge: bool = False,
-) -> None:
-    """
-    Bulk equivalent of set_permissions_for_object: applies the same
-    permission changes to every object identified by `pks` at once.
-
-    Takes a model + pks (rather than model instances) deliberately -- the
-    permission rows built below only ever need `pk`, `content_type`, and
-    identity ids, so callers shouldn't have to fetch full rows (with every
-    other field) just to hand them to this function.
-
-    Deliberately does not use guardian's queryset/list-aware assign_perm:
-    passing a list as the object routes to bulk_assign_perm, which skips
-    creating a direct permission row for anyone who already has the
-    permission via ANY group membership (it checks
-    ObjectPermissionChecker.has_perm, which is group-inheritance-aware) --
-    unlike the single-object assign_perm this replaces, which always
-    ensures a direct row via get_or_create regardless of group-derived
-    access. Losing that guarantee would mean a later revocation of the
-    group's grant silently strips access an admin explicitly asked to be
-    direct. Bulk-creating rows straight against the permission models
-    instead (see _apply_bulk_permission_entry) preserves the original
-    always-create-a-direct-row semantics while still batching every object
-    and every identity into one query per action, rather than one query per
-    (object, user) pair.
-    """
-    object_pks = [str(pk) for pk in pks]
-    if not object_pks:
-        return
-
-    model_name = model.__name__.lower()
-    ctype = ContentType.objects.get_for_model(model)
-
-    for action, entry in permissions.items():
-        codename = f"{action}_{model_name}"
-        implied_codenames = {codename}
-        if action == "change":
-            # change gives view too
-            implied_codenames.add(f"view_{model_name}")
-
-        # Resolved once per action (not once per users/groups branch) and
-        # shared between both below -- also where an unrecognized action
-        # name (see _resolve_permissions) is caught.
-        permission_objs = (
-            _resolve_permissions(implied_codenames, ctype)
-            if "users" in entry or "groups" in entry
-            else []
-        )
-
-        if "users" in entry:
-            _apply_bulk_permission_entry(
-                perm_model=UserObjectPermission,
-                identity_model=User,
-                identity_field="user",
-                ids=entry["users"],
-                codename=codename,
-                permission_objs=permission_objs,
-                ctype=ctype,
-                object_pks=object_pks,
-                merge=merge,
-            )
-
-        if "groups" in entry:
-            _apply_bulk_permission_entry(
-                perm_model=GroupObjectPermission,
-                identity_model=Group,
-                identity_field="group",
-                ids=entry["groups"],
-                codename=codename,
-                permission_objs=permission_objs,
-                ctype=ctype,
-                object_pks=object_pks,
-                merge=merge,
-            )
-
-
 def permitted_object_ids(
     user: User | None,
     model: type[Model],

src/documents/serialisers.py

@@ -1,10 +1,8 @@
 from __future__ import annotations
 
-import contextlib
 import logging
 import math
 import re
-from collections.abc import Iterable
 from datetime import datetime
 from datetime import timedelta
 from decimal import Decimal
@@ -26,7 +24,6 @@ from django.core.validators import MaxValueValidator
 from django.core.validators import MinValueValidator
 from django.core.validators import RegexValidator
 from django.core.validators import integer_validator
-from django.db import DataError
 from django.db.models import Count
 from django.db.models import Q
 from django.db.models.functions import Lower
@@ -40,12 +37,12 @@ from django.utils.timezone import make_aware
 from django.utils.translation import gettext as _
 from drf_spectacular.utils import extend_schema_field
 from drf_spectacular.utils import extend_schema_serializer
+from drf_writable_nested.serializers import NestedUpdateMixin
 from guardian.core import ObjectPermissionChecker
 from guardian.shortcuts import get_users_with_perms
 from guardian.utils import get_group_obj_perms_model
 from guardian.utils import get_user_obj_perms_model
 from rest_framework import fields
-from rest_framework import relations
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.fields import SerializerMethodField
@@ -745,100 +742,22 @@ class TagSerializer(MatchingModelSerializer, OwnedObjectSerializer):
         return super().validate(attrs)
 
 
-class _BatchingManyRelatedField(serializers.ManyRelatedField):
-    """
-    `ManyRelatedField.to_internal_value` resolves each id in the submitted
-    list with its own `child_relation.to_internal_value()` call -- one query
-    per item on every PATCH/PUT that sets a `many=True` relation field.
-    Batch-resolve them instead, falling back to the child relation's normal
-    (query-per-item) validation for anything that isn't a plausible int pk,
-    so bad input still gets the usual DRF validation error rather than being
-    silently dropped.
-    """
-
-    @staticmethod
-    def _normalize_pk(item) -> int | None:
-        # Excludes bool: DRF's own PrimaryKeyRelatedField rejects it too
-        # (True == 1 would otherwise silently match pk 1).
-        if isinstance(item, bool):
-            return None
-        try:
-            return int(item)
-        except (TypeError, ValueError):
-            return None
-
-    def to_internal_value(self, data):
-        if isinstance(data, str) or not hasattr(data, "__iter__"):
-            self.fail("not_a_list", input_type=type(data).__name__)
-        if not self.allow_empty and len(data) == 0:
-            self.fail("empty")
-
-        item_pks = [(item, self._normalize_pk(item)) for item in data]
-        candidate_pks = {pk for _, pk in item_pks if pk is not None}
-
-        # Django's IntegerFieldOverflow guard (-> EmptyResultSet, i.e. no
-        # match) only covers exact/gt/gte/lt/lte lookups, not `in` -- an
-        # out-of-range int in `pk__in=` reaches the DB driver as-is and
-        # raises OverflowError (SQLite) / DataError (Postgres) instead of
-        # cleanly matching nothing. The per-item `exact`-lookup fallback
-        # below IS covered, so on that failure just skip the batch and let
-        # every item resolve individually -- each still costs one query,
-        # but reports the normal validation error instead of a raw 500.
-        try:
-            resolved_by_pk = {
-                obj.pk: obj
-                for obj in self.child_relation.get_queryset().filter(
-                    pk__in=candidate_pks,
-                )
-            }
-        except (OverflowError, DataError):
-            resolved_by_pk = {}
-
-        result = []
-        for item, pk in item_pks:
-            obj = resolved_by_pk.get(pk) if pk is not None else None
-            result.append(
-                obj if obj is not None else self.child_relation.to_internal_value(item),
-            )
-        return result
-
-
-class BatchResolvingPrimaryKeyRelatedField(serializers.PrimaryKeyRelatedField):
-    """
-    A PrimaryKeyRelatedField whose `many=True` form (a DRF ManyRelatedField)
-    resolves all submitted ids with one batched query instead of one query
-    per id. Subclasses only need to implement `get_queryset()` as usual --
-    only `TagsField` is used with `many=True` today, but this is the base
-    for all four so the fix isn't tag-specific: if a future PR puts
-    `many=True` on correspondent/document_type/storage_path, it inherits the
-    same batching instead of reintroducing this as a new bug to rediscover.
-    """
-
-    @classmethod
-    def many_init(cls, *args, **kwargs):
-        list_kwargs = {"child_relation": cls(*args, **kwargs)}
-        for key, value in kwargs.items():
-            if key in relations.MANY_RELATION_KWARGS:
-                list_kwargs[key] = value
-        return _BatchingManyRelatedField(**list_kwargs)
-
-
-class CorrespondentField(BatchResolvingPrimaryKeyRelatedField[Correspondent]):
+class CorrespondentField(serializers.PrimaryKeyRelatedField[Correspondent]):
     def get_queryset(self):
         return Correspondent.objects.all()
 
 
-class TagsField(BatchResolvingPrimaryKeyRelatedField[Tag]):
+class TagsField(serializers.PrimaryKeyRelatedField[Tag]):
     def get_queryset(self):
         return Tag.objects.all()
 
 
-class DocumentTypeField(BatchResolvingPrimaryKeyRelatedField[DocumentType]):
+class DocumentTypeField(serializers.PrimaryKeyRelatedField[DocumentType]):
     def get_queryset(self):
         return DocumentType.objects.all()
 
 
-class StoragePathField(BatchResolvingPrimaryKeyRelatedField[StoragePath]):
+class StoragePathField(serializers.PrimaryKeyRelatedField[StoragePath]):
     def get_queryset(self):
         return StoragePath.objects.all()
 
@@ -957,106 +876,8 @@ def validate_documentlink_targets(user, doc_ids):
         )
 
 
-# A CustomField lookup cache scoped to a single field/serializer instance
-# only helps within that one instance's own validation pass. Several call
-# sites, though, build more than one CustomFieldInstanceSerializer (or its
-# CustomFieldInstanceListSerializer/field) for the same request and pass
-# each of them `context=self.context` -- the *same* dict object, not a
-# copy -- e.g. bulk-edit's _validate_custom_field_values() constructing a
-# fresh CustomFieldInstanceSerializer per submitted field. That context
-# dict is already request-scoped (DRF builds it fresh per request via
-# get_serializer_context()), so stashing the resolved CustomField objects
-# there -- rather than in some new global/thread-local cache -- lets every
-# one of those separately-instantiated serializers reuse them for free
-# while staying entirely within DRF's existing, already-request-scoped
-# machinery.
-_CUSTOM_FIELD_CONTEXT_CACHE_KEY = "_custom_field_lookup_cache"
-
-
-class _CachingCustomFieldPrimaryKeyField(serializers.PrimaryKeyRelatedField):
-    """
-    Resolves CustomField ids with as few queries as possible: a per-instance
-    cache for repeat lookups on this exact field instance, backed by a
-    shared cache on the serializer context (see _CUSTOM_FIELD_CONTEXT_CACHE_KEY
-    above) so later, separately-instantiated fields for the same request
-    reuse what was already resolved instead of re-querying.
-    """
-
-    def __init__(self, **kwargs: Any) -> None:
-        super().__init__(**kwargs)
-        self._cache: dict[int, CustomField] = {}
-
-    def _shared_cache(self) -> dict[int, CustomField]:
-        return self.context.setdefault(_CUSTOM_FIELD_CONTEXT_CACHE_KEY, {})
-
-    @staticmethod
-    def _normalize_pk(data: Any) -> int | None:
-        """
-        Returns `data` coerced to the int a valid CustomField pk would be,
-        or None if `data` isn't a plausible pk (wrong type, unhashable,
-        non-numeric, or a bool -- DRF itself rejects bools as pks since
-        `True == 1` would otherwise silently match). None tells callers to
-        leave `data` alone and let `super().to_internal_value()` report the
-        normal validation error instead of touching the cache/queryset with
-        it directly.
-        """
-        if isinstance(data, bool):
-            return None
-        try:
-            return int(data)
-        except (TypeError, ValueError):
-            return None
-
-    def prefetch(self, ids: Iterable[Any]) -> None:
-        shared_cache = self._shared_cache()
-        candidates = {pk for i in ids if (pk := self._normalize_pk(i)) is not None}
-        missing = {
-            i for i in candidates if i not in self._cache and i not in shared_cache
-        }
-        if missing:
-            for obj in self.get_queryset().filter(pk__in=missing):
-                shared_cache[obj.pk] = obj
-        for i in candidates:
-            obj = shared_cache.get(i)
-            if obj is not None:
-                self._cache[i] = obj
-
-    def to_internal_value(self, data: Any) -> CustomField:
-        pk = self._normalize_pk(data)
-        if pk is None:
-            return super().to_internal_value(data)
-        if pk in self._cache:
-            return self._cache[pk]
-        shared_cache = self._shared_cache()
-        if pk in shared_cache:
-            obj = shared_cache[pk]
-            self._cache[pk] = obj
-            return obj
-        obj: CustomField = super().to_internal_value(data)
-        self._cache[obj.pk] = obj
-        shared_cache[obj.pk] = obj
-        return obj
-
-
-class CustomFieldInstanceListSerializer(serializers.ListSerializer):
-    def to_internal_value(self, data: Any) -> list[Any]:
-        if isinstance(data, list):
-            field_ids = []
-            for item in data:
-                if not isinstance(item, dict) or "field" not in item:
-                    continue
-                try:
-                    hash(item["field"])
-                except TypeError:
-                    continue
-                field_ids.append(item["field"])
-            if field_ids:
-                self.child.fields["field"].prefetch(field_ids)
-        return super().to_internal_value(data)
-
-
 class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInstance]):
-    field = _CachingCustomFieldPrimaryKeyField(queryset=CustomField.objects.all())
+    field = serializers.PrimaryKeyRelatedField(queryset=CustomField.objects.all())
     value = ReadWriteSerializerMethodField(allow_null=True)
 
     def create(self, validated_data):
@@ -1157,7 +978,6 @@ class CustomFieldInstanceSerializer(serializers.ModelSerializer[CustomFieldInsta
 
     class Meta:
         model = CustomFieldInstance
-        list_serializer_class = CustomFieldInstanceListSerializer
         fields = [
             "value",
             "field",
@@ -1223,6 +1043,7 @@ class DocumentVersionInfoSerializer(serializers.Serializer[_DocumentVersionInfo]
 )
 class DocumentSerializer(
     OwnedObjectSerializer,
+    NestedUpdateMixin,
     DocumentUpdateFieldsModelSerializer,
 ):
     correspondent = CorrespondentField(allow_null=True)
@@ -1437,60 +1258,16 @@ class DocumentSerializer(
                     if tag not in inbox_tags_not_being_added
                 ]
 
-        custom_fields_data = validated_data.pop("custom_fields", None)
-
-        actor_context = (
-            set_actor(self.user)
-            if settings.AUDIT_LOG_ENABLED
-            else contextlib.nullcontext()
-        )
-        with actor_context:
+        if settings.AUDIT_LOG_ENABLED:
+            with set_actor(self.user):
+                super().update(instance, validated_data)
+        else:
             super().update(instance, validated_data)
-            if custom_fields_data is not None:
-                self._sync_custom_fields(instance, custom_fields_data)
 
+        # hard delete custom field instances that were soft deleted
+        CustomFieldInstance.deleted_objects.filter(document=instance).delete()
         return instance
 
-    def _sync_custom_fields(
-        self,
-        instance: Document,
-        custom_fields_data: list[dict],
-    ) -> None:
-        """
-        Create/update a CustomFieldInstance for every (field, value) pair in
-        custom_fields_data, then hard-delete any of the document's existing
-        instances whose field wasn't included.
-
-        Replaces drf-writable-nested's generic
-        update_or_create_reverse_relations()/delete_reverse_relations_if_need():
-        that machinery always matched submitted items by an instance "id"
-        this client payload never sends, so its own pk-matching never did
-        anything for this field -- the real upsert semantics were always
-        CustomFieldInstanceSerializer.create()'s update_or_create() below.
-
-        On a partial (PATCH) update, DRF skips the "value" field's required
-        check on the first validation pass because it's absent entirely from
-        the payload item, not merely null. drf-writable-nested happened to
-        re-enforce that check itself, by re-validating each item against a
-        freshly built, non-partial child serializer before saving. Reproduce
-        that specific guarantee explicitly here, since CustomFieldInstance's
-        "value" is not optional.
-        """
-        for item in custom_fields_data:
-            if "value" not in item:
-                raise serializers.ValidationError(
-                    {"custom_fields": [{"value": ["This field is required."]}]},
-                )
-
-        kept_field_ids: set[int] = set()
-        serializer = CustomFieldInstanceSerializer()
-        for item in custom_fields_data:
-            kept_field_ids.add(item["field"].pk)
-            serializer.create({**item, "document": instance})
-        CustomFieldInstance.objects.filter(document=instance).exclude(
-            field_id__in=kept_field_ids,
-        ).hard_delete()
-
     def __init__(self, *args, **kwargs) -> None:
         self.truncate_content = kwargs.pop("truncate_content", False)
 

src/documents/tests/test_api_custom_fields.py

@@ -5,9 +5,7 @@ from unittest.mock import ANY
 
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
-from django.db import connection
 from django.test import override_settings
-from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
 from rest_framework import status
 from rest_framework.test import APITestCase
@@ -15,9 +13,6 @@ from rest_framework.test import APITestCase
 from documents.models import CustomField
 from documents.models import CustomFieldInstance
 from documents.models import Document
-from documents.serialisers import CustomFieldInstanceSerializer
-from documents.serialisers import DocumentSerializer
-from documents.tests.factories import DocumentFactory
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -535,137 +530,6 @@ class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
         doc.refresh_from_db()
         self.assertEqual(len(doc.custom_fields.all()), 10)
 
-    def test_document_serializer_custom_fields_validation_batches_field_lookup(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - A document is being validated with several custom field values
-              at once (as happens on every PATCH/PUT/POST)
-        WHEN:
-            - The serializer is validated
-        THEN:
-            - The referenced CustomField objects are resolved with a single
-              query, not one query per custom field
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-        custom_fields = [
-            CustomField.objects.create(
-                name=f"Test Custom Field {i}",
-                data_type=CustomField.FieldDataType.STRING,
-            )
-            for i in range(5)
-        ]
-
-        serializer = DocumentSerializer(
-            doc,
-            data={
-                "custom_fields": [
-                    {"field": custom_field.id, "value": "test value"}
-                    for custom_field in custom_fields
-                ],
-            },
-            partial=True,
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            self.assertTrue(serializer.is_valid(), serializer.errors)
-
-        custom_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            len(custom_field_lookups),
-            1,
-            "Expected a single batched query to resolve the custom fields, "
-            f"got {len(custom_field_lookups)}: {custom_field_lookups}",
-        )
-
-    def test_custom_field_lookup_reuses_shared_context_cache(self) -> None:
-        """
-        GIVEN:
-            - A CustomField has already been resolved once, by a serializer
-              sharing a given `context` dict
-        WHEN:
-            - A second, separately-instantiated CustomFieldInstanceSerializer
-              validates the same field id, sharing that same context
-              (this mirrors what real call sites do, e.g. bulk-edit's
-              validate_custom_fields()/_validate_custom_field_values()
-              constructing a fresh CustomFieldInstanceSerializer per
-              submitted field, all sharing the outer serializer's context)
-        THEN:
-            - No additional query is issued to resolve the CustomField
-        """
-        custom_field = CustomField.objects.create(
-            name="Test Custom Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-
-        context: dict = {}
-        first_pass = CustomFieldInstanceSerializer(
-            data={"field": custom_field.id, "value": "a"},
-            context=context,
-        )
-        self.assertTrue(first_pass.is_valid(), first_pass.errors)
-
-        second_pass = CustomFieldInstanceSerializer(
-            data={"field": custom_field.id, "value": "b"},
-            context=context,
-        )
-        with CaptureQueriesContext(connection) as ctx:
-            self.assertTrue(second_pass.is_valid(), second_pass.errors)
-
-        custom_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            len(custom_field_lookups),
-            0,
-            "Expected the second, separately-instantiated serializer to reuse "
-            f"the already-resolved CustomField, got: {custom_field_lookups}",
-        )
-
-    def test_custom_field_validation_rejects_malformed_field_value(self) -> None:
-        """
-        GIVEN:
-            - A document is being validated with a malformed custom_fields
-              entry whose "field" value is neither a valid CustomField id
-              nor a type DRF's own PrimaryKeyRelatedField can safely reject
-              on its own (unhashable, or a non-numeric scalar)
-        WHEN:
-            - The serializer is validated
-        THEN:
-            - A normal validation error is raised, not an unhandled
-              TypeError/ValueError escaping past DRF's validation layer
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-
-        bad_field_values = {
-            "unhashable-list": [],
-            "unhashable-dict": {},
-            "non-numeric-scalar": "abc",
-        }
-        for case_id, bad_field_value in bad_field_values.items():
-            with self.subTest(case_id):
-                serializer = DocumentSerializer(
-                    doc,
-                    data={
-                        "custom_fields": [
-                            {"field": bad_field_value, "value": "test value"},
-                        ],
-                    },
-                    partial=True,
-                )
-
-                self.assertFalse(serializer.is_valid())
-                self.assertIn("custom_fields", serializer.errors)
-
     def test_change_custom_field_instance_value(self) -> None:
         """
         GIVEN:
@@ -796,72 +660,6 @@ class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
         assert _cf_4 is not None
         self.assertEqual(_cf_4.value, date_value)
 
-    def test_delete_custom_field_instance_is_hard_deleted_not_soft_deleted(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - A document has two custom field instances
-        WHEN:
-            - A PATCH request updates custom_fields to omit one of them
-        THEN:
-            - The omitted instance is immediately hard-deleted: it is gone
-              from both the default manager and the soft-deleted manager,
-              not left in a soft-deleted, not-yet-purged state
-        """
-        doc = Document.objects.create(
-            title="WOW",
-            content="the content",
-            checksum="123-hard-delete",
-            mime_type="application/pdf",
-        )
-        kept_field = CustomField.objects.create(
-            name="Kept Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-        removed_field = CustomField.objects.create(
-            name="Removed Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-
-        resp = self.client.patch(
-            f"/api/documents/{doc.id}/",
-            data={
-                "custom_fields": [
-                    {"field": kept_field.id, "value": "keep me"},
-                    {"field": removed_field.id, "value": "remove me"},
-                ],
-            },
-            format="json",
-        )
-        self.assertEqual(resp.status_code, status.HTTP_200_OK)
-        self.assertEqual(CustomFieldInstance.objects.count(), 2)
-
-        resp = self.client.patch(
-            f"/api/documents/{doc.id}/",
-            data={
-                "custom_fields": [
-                    {"field": kept_field.id, "value": "keep me"},
-                ],
-            },
-            format="json",
-        )
-        self.assertEqual(resp.status_code, status.HTTP_200_OK)
-
-        self.assertEqual(CustomFieldInstance.objects.count(), 1)
-        self.assertEqual(
-            CustomFieldInstance.deleted_objects.filter(field=removed_field).count(),
-            0,
-            "Removed custom field instance should be hard-deleted, not left "
-            "soft-deleted",
-        )
-        self.assertEqual(
-            CustomFieldInstance.global_objects.filter(field=removed_field).count(),
-            0,
-            "Removed custom field instance should not exist at all, even in "
-            "the all-rows manager",
-        )
-
     def test_custom_field_validation(self) -> None:
         """
         GIVEN:
@@ -1553,81 +1351,6 @@ class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
         results = response.data["results"]
         self.assertEqual(results[0]["document_count"], 0)
 
-    def test_document_update_custom_fields_sync_query_count(self) -> None:
-        """
-        GIVEN:
-            - A document already has 3 custom field instances attached
-        WHEN:
-            - A PATCH request updates 2 of them, adds 1 new one, and omits
-              the 3rd (which should be deleted)
-        THEN:
-            - The omitted custom field is removed via exactly one
-              delete-shaped query, not drf-writable-nested's old two-step
-              soft-delete-then-hard-delete-after pattern (two delete
-              passes)
-        """
-        doc = Document.objects.create(
-            title="WOW",
-            content="the content",
-            checksum="123-sync-count",
-            mime_type="application/pdf",
-        )
-        fields = [
-            CustomField.objects.create(
-                name=f"Sync Field {i}",
-                data_type=CustomField.FieldDataType.STRING,
-            )
-            for i in range(4)
-        ]
-        # Attach the first 3 up front; the 4th is added in the PATCH below,
-        # and the 3rd is omitted (so it should be deleted).
-        for field in fields[:3]:
-            CustomFieldInstance.objects.create(
-                document=doc,
-                field=field,
-                value_text="initial",
-            )
-        self.assertEqual(CustomFieldInstance.objects.count(), 3)
-
-        with CaptureQueriesContext(connection) as ctx:
-            resp = self.client.patch(
-                f"/api/documents/{doc.id}/",
-                data={
-                    "custom_fields": [
-                        {"field": fields[0].id, "value": "updated 0"},
-                        {"field": fields[1].id, "value": "updated 1"},
-                        {"field": fields[3].id, "value": "new 3"},
-                    ],
-                },
-                format="json",
-            )
-        self.assertEqual(resp.status_code, status.HTTP_200_OK)
-
-        delete_queries = [
-            q
-            for q in ctx.captured_queries
-            if 'DELETE FROM "documents_customfieldinstance"' in q["sql"]
-            or (
-                'UPDATE "documents_customfieldinstance"' in q["sql"]
-                and '"deleted_at"' in q["sql"]
-                and '"deleted_at" = NULL' not in q["sql"]
-            )
-        ]
-        self.assertEqual(
-            len(delete_queries),
-            1,
-            "Expected exactly one delete/soft-delete-marking query for the "
-            f"omitted custom field, got {len(delete_queries)}: {delete_queries}",
-        )
-
-        self.assertEqual(CustomFieldInstance.objects.count(), 3)
-        doc.refresh_from_db()
-        values = {cfi.field_id: cfi.value for cfi in doc.custom_fields.all()}
-        self.assertEqual(values[fields[0].id], "updated 0")
-        self.assertEqual(values[fields[1].id], "updated 1")
-        self.assertEqual(values[fields[3].id], "new 3")
-        self.assertNotIn(fields[2].id, values)
-
     def test_patch_document_invalid_date_custom_field_returns_validation_error(
         self,
     ) -> None:

src/documents/tests/test_api_document_versions.py

@@ -11,6 +11,7 @@ from django.contrib.auth.models import User
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import FieldError
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase as DjangoTestCase
 from django.utils import timezone
 from rest_framework import status
 from rest_framework.test import APITestCase
@@ -21,6 +22,7 @@ from documents.filters import TitleContentFilter
 from documents.models import Document
 from documents.tests.utils import DirectoriesMixin
 from documents.tests.utils import read_streaming_response
+from documents.views import DocumentSelectionMixin
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -923,3 +925,36 @@ class TestVersionAwareFilters(TestCase):
 
         self.assertIs(result, queryset)
         queryset.filter.assert_not_called()
+
+
+class TestBulkSelectionExcludesVersions(DjangoTestCase):
+    def test_select_all_matching_does_not_select_version_documents(self) -> None:
+        """
+        "Select all matching" reconstructs the document list, which never
+        contains version documents as rows of their own.
+        """
+        user = User.objects.create_superuser(username="bulk_versions")
+        root = Document.objects.create(
+            title="shared-title root",
+            checksum="bulk-root",
+            mime_type="application/pdf",
+            content="root",
+        )
+        Document.objects.create(
+            title="shared-title version",
+            checksum="bulk-version",
+            mime_type="application/pdf",
+            root_document=root,
+            version_index=1,
+            content="version",
+        )
+
+        selected = DocumentSelectionMixin()._resolve_document_ids(
+            user=user,
+            validated_data={
+                "all": True,
+                "filters": {"title__icontains": "shared-title"},
+            },
+        )
+
+        self.assertEqual(selected, [root.id])

src/documents/tests/test_api_documents.py

@@ -1,6 +1,5 @@
 import datetime
 import json
-import re
 import shutil
 import tempfile
 import uuid
@@ -22,9 +21,7 @@ from django.core import mail
 from django.core.cache import cache
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.db import DataError
-from django.db import connection
 from django.test import override_settings
-from django.test.utils import CaptureQueriesContext
 from django.utils import timezone
 from guardian.shortcuts import assign_perm
 from rest_framework import status
@@ -255,82 +252,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
         doc.refresh_from_db()
         self.assertEqual(doc.created, date(2023, 6, 28))
 
-    def test_document_update_tags_batches_tag_lookup(self) -> None:
-        """
-        GIVEN:
-            - A document is being updated with several tags at once
-        WHEN:
-            - API PATCH request is made setting the document's tags
-        THEN:
-            - The referenced Tag objects are resolved with a single batched
-              query, not one query per tag
-        """
-        doc = Document.objects.create(
-            title="none",
-            checksum="123",
-            mime_type="application/pdf",
-        )
-        tags = [TagFactory() for _ in range(8)]
-
-        with CaptureQueriesContext(connection) as ctx:
-            response = self.client.patch(
-                f"/api/documents/{doc.pk}/",
-                {"tags": [t.id for t in tags]},
-                format="json",
-            )
-        self.assertEqual(response.status_code, status.HTTP_200_OK)
-
-        # Match `"documents_tag"."id" = <literal>` (a single-row WHERE lookup)
-        # but not the same substring appearing as a JOIN's ON condition
-        # (`"documents_tag"."id" = "documents_document_tags"."tag_id"`),
-        # which is a legitimate, unrelated response-serialization query.
-        single_tag_lookup_re = re.compile(r'"documents_tag"\."id" = \d')
-        single_tag_lookups = [
-            q for q in ctx.captured_queries if single_tag_lookup_re.search(q["sql"])
-        ]
-        self.assertEqual(
-            len(single_tag_lookups),
-            0,
-            "Expected tags to be resolved with a batched query, not "
-            f"per-tag lookups, got: {single_tag_lookups}",
-        )
-
-        doc.refresh_from_db()
-        self.assertCountEqual(
-            doc.tags.values_list("id", flat=True),
-            [t.id for t in tags],
-        )
-
-    def test_document_update_tags_rejects_out_of_range_id(self) -> None:
-        """
-        GIVEN:
-            - A document is being updated with a tag id too large for the
-              database's integer column
-        WHEN:
-            - API PATCH request is made setting the document's tags
-        THEN:
-            - A normal 400 validation error is returned, not an unhandled
-              OverflowError/DataError escaping as a 500
-
-        Django's IntegerFieldOverflow guard converts an out-of-range int
-        into a clean "no match" for exact/gt/gte/lt/lte lookups, but not for
-        `in` -- the batched tag resolution uses `pk__in=`, so this has to be
-        guarded explicitly rather than relying on Django to do it.
-        """
-        doc = Document.objects.create(
-            title="none",
-            checksum="123",
-            mime_type="application/pdf",
-        )
-
-        response = self.client.patch(
-            f"/api/documents/{doc.pk}/",
-            {"tags": [99999999999999999999999999999]},
-            format="json",
-        )
-
-        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
-
     def test_document_update_legacy_created_format(self) -> None:
         """
         GIVEN:

src/documents/tests/test_api_objects.py

@@ -2,15 +2,10 @@ import datetime
 import json
 from unittest import mock
 
-from django.contrib.auth.models import Group
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
-from django.db import connection
 from django.test import override_settings
-from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
-from guardian.shortcuts import get_groups_with_perms
-from guardian.shortcuts import get_users_with_perms
 from rest_framework import status
 from rest_framework.test import APITestCase
 
@@ -820,59 +815,6 @@ class TestBulkEditObjects(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(StoragePath.objects.count(), 0)
 
-    def test_bulk_objects_set_permissions_query_count_independent_of_object_count(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - Many tags are being bulk-edited to set permissions at once
-        WHEN:
-            - bulk_edit_objects API endpoint is called with set_permissions
-              operation over a small batch vs. a much larger one
-        THEN:
-            - The number of queries issued is the same either way -- each
-              user/group is applied across all tags with one batched call,
-              not one call per (tag, user) pair
-        """
-        group1 = Group.objects.create(name="perm-group")
-        permissions = {
-            "view": {"users": [self.user1.id, self.user2.id], "groups": [group1.id]},
-            "change": {"users": [self.user1.id], "groups": [group1.id]},
-        }
-
-        def run_with_n_tags(n: int) -> int:
-            tags = [Tag.objects.create(name=f"perm-tag-{n}-{i}") for i in range(n)]
-            with CaptureQueriesContext(connection) as ctx:
-                response = self.client.post(
-                    "/api/bulk_edit_objects/",
-                    json.dumps(
-                        {
-                            "objects": [t.id for t in tags],
-                            "object_type": "tags",
-                            "operation": "set_permissions",
-                            "permissions": permissions,
-                            "merge": False,
-                        },
-                    ),
-                    content_type="application/json",
-                )
-            self.assertEqual(response.status_code, status.HTTP_200_OK)
-            for tag in tags:
-                self.assertEqual(get_users_with_perms(tag).count(), 2)
-                self.assertEqual(get_groups_with_perms(tag).count(), 1)
-            return len(ctx.captured_queries)
-
-        small_batch_queries = run_with_n_tags(5)
-        large_batch_queries = run_with_n_tags(50)
-
-        self.assertEqual(
-            small_batch_queries,
-            large_batch_queries,
-            "Expected the same query count regardless of tag count, got "
-            f"{small_batch_queries} queries for 5 tags vs. "
-            f"{large_batch_queries} for 50",
-        )
-
     def test_bulk_objects_delete_all_filtered(self) -> None:
         """
         GIVEN:

src/documents/tests/test_bulk_edit.py

@@ -5,11 +5,8 @@ from unittest import mock
 
 import pikepdf
 from django.contrib.auth.models import Group
-from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
-from django.db import connection
 from django.test import TestCase
-from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
 from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_users_with_perms
@@ -22,7 +19,6 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import set_permissions_for_objects
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -348,100 +344,6 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         assert _cf_3 is not None
         self.assertNotIn(self.doc3.id, _cf_3.value)
 
-    def test_modify_custom_fields_batches_field_lookup(self) -> None:
-        """
-        GIVEN:
-            - Several documents are being bulk-edited to add several custom
-              fields at once
-        WHEN:
-            - modify_custom_fields runs
-        THEN:
-            - Each CustomField is resolved with one batched query total, not
-              once per (field, document) pair
-        """
-        docs = [
-            Document.objects.create(checksum=f"batch-{i}", title=f"batch-{i}")
-            for i in range(6)
-        ]
-        fields = [
-            CustomField.objects.create(
-                name=f"Batch Field {i}",
-                data_type=CustomField.FieldDataType.STRING,
-            )
-            for i in range(4)
-        ]
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields=[field.id for field in fields],
-                remove_custom_fields=[],
-            )
-
-        field_lookups = [
-            q
-            for q in ctx.captured_queries
-            if 'FROM "documents_customfield"' in q["sql"]
-        ]
-        self.assertEqual(
-            len(field_lookups),
-            1,
-            "Expected a single batched query to resolve the custom fields, "
-            f"got {len(field_lookups)}: {field_lookups}",
-        )
-
-        for doc in docs:
-            self.assertEqual(doc.custom_fields.count(), len(fields))
-
-    def test_modify_custom_fields_batches_document_lookup_for_documentlink(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - Several documents are being bulk-edited to add a DOCUMENTLINK
-              custom field at once
-        WHEN:
-            - modify_custom_fields runs
-        THEN:
-            - The Document rows needed to reflect the symmetrical links are
-              resolved with one batched query total, not once per document
-        """
-        docs = [
-            Document.objects.create(checksum=f"link-{i}", title=f"link-{i}")
-            for i in range(6)
-        ]
-        target = Document.objects.create(checksum="link-target", title="link-target")
-        doclink_field = CustomField.objects.create(
-            name="Related",
-            data_type=CustomField.FieldDataType.DOCUMENTLINK,
-        )
-
-        with CaptureQueriesContext(connection) as ctx:
-            bulk_edit.modify_custom_fields(
-                [doc.id for doc in docs],
-                add_custom_fields={doclink_field.id: [target.id]},
-                remove_custom_fields=[],
-            )
-
-        single_document_lookups = [
-            q
-            for q in ctx.captured_queries
-            if 'FROM "documents_document"' in q["sql"]
-            and '"documents_document"."id" = ' in q["sql"]
-        ]
-        self.assertEqual(
-            len(single_document_lookups),
-            0,
-            "Expected document rows to come from a batched query, not "
-            f"per-document lookups, got: {single_document_lookups}",
-        )
-
-        for doc in docs:
-            self.assertEqual(
-                doc.custom_fields.get(field=doclink_field).value,
-                [target.id],
-            )
-
     def test_modify_custom_fields_doclink_self_link(self) -> None:
         """
         GIVEN:
@@ -608,132 +510,6 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         )
         self.assertEqual(groups_with_perms.count(), 2)
 
-    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
-    def test_set_permissions_query_count_independent_of_document_count(
-        self,
-        m,
-    ) -> None:
-        """
-        GIVEN:
-            - Many documents are being bulk-edited to set permissions at once
-        WHEN:
-            - set_permissions runs over a small batch vs. a much larger one
-        THEN:
-            - The number of queries issued is the same either way -- each
-              user/group is applied across all documents with one batched
-              call, not one call per (document, user) pair
-        """
-        permissions = {
-            "view": {
-                "users": [self.user1.id, self.user2.id],
-                "groups": [self.group2.id],
-            },
-            "change": {
-                "users": [self.user1.id],
-                "groups": [self.group2.id],
-            },
-        }
-
-        def run_with_n_documents(n: int) -> int:
-            docs = [
-                Document.objects.create(checksum=f"perm-{n}-{i}", title=f"perm-{n}-{i}")
-                for i in range(n)
-            ]
-            with CaptureQueriesContext(connection) as ctx:
-                bulk_edit.set_permissions(
-                    [doc.id for doc in docs],
-                    set_permissions=permissions,
-                    owner=self.owner,
-                    merge=False,
-                )
-            for doc in docs:
-                self.assertEqual(get_users_with_perms(doc).count(), 2)
-                self.assertEqual(get_groups_with_perms(doc).count(), 1)
-            return len(ctx.captured_queries)
-
-        small_batch_queries = run_with_n_documents(5)
-        large_batch_queries = run_with_n_documents(50)
-
-        self.assertEqual(
-            small_batch_queries,
-            large_batch_queries,
-            "Expected the same query count regardless of document count, got "
-            f"{small_batch_queries} queries for 5 documents vs. "
-            f"{large_batch_queries} for 50",
-        )
-
-    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
-    def test_set_permissions_grants_direct_perm_even_if_already_granted_via_group(
-        self,
-        m,
-    ) -> None:
-        """
-        GIVEN:
-            - A user already has view access to a document via group
-              membership, with no direct grant of their own
-        WHEN:
-            - set_permissions explicitly grants that same user direct view
-              access via bulk_edit
-        THEN:
-            - A direct permission grant is created for the user, not skipped
-              because they already have equivalent access via the group
-
-        Regression test: guardian's queryset-aware assign_perm() (routed to
-        when the target is a list/queryset) skips creating a direct row for
-        anyone whose ObjectPermissionChecker.has_perm() already returns True
-        -- which includes group-derived access. The single-object assign_perm
-        this bulk path replaces has no such check; it always ensures a
-        direct row via get_or_create. Losing that guarantee would mean
-        revoking the group's grant later silently strips access that was
-        supposed to be explicit.
-        """
-        self.doc1.owner = self.user1
-        self.doc1.save()
-        self.user1.groups.add(self.group1)
-        assign_perm("view_document", self.group1, self.doc1)
-
-        bulk_edit.set_permissions(
-            [self.doc1.id],
-            set_permissions={
-                "view": {"users": [self.user1.id], "groups": []},
-            },
-            merge=True,
-        )
-
-        direct_users = get_users_with_perms(
-            self.doc1,
-            only_with_perms_in=["view_document"],
-            with_group_users=False,
-        )
-        self.assertIn(self.user1, direct_users)
-
-    def test_set_permissions_for_objects_raises_for_unknown_action(self) -> None:
-        """
-        GIVEN:
-            - An unrecognized permission action name with users to grant it
-              to
-        WHEN:
-            - set_permissions_for_objects is called
-        THEN:
-            - Permission.DoesNotExist is raised, not a silent no-op
-
-        Regression test: the endpoint that calls this
-        (BulkEditObjectPermissionsView) never actually validates action
-        names against the raw client-supplied permissions dict --
-        BulkEditObjectsSerializer._validate_permissions calls
-        validate_set_permissions() only for its side-effecting user/group id
-        checks and discards the filtered dict it returns -- so a bogus
-        action key reaches this function as-is. Resolving the Permission via
-        a bare `.filter()` (which returns empty instead of raising) would
-        silently drop the grant and report success.
-        """
-        with self.assertRaises(Permission.DoesNotExist):
-            set_permissions_for_objects(
-                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},
-                Document,
-                [self.doc1.pk],
-            )
-
     @mock.patch("documents.models.Document.delete")
     def test_delete_documents_old_uuid_field(self, m) -> None:
         m.side_effect = Exception("Data too long for column 'transaction_id' at row 1")

src/documents/tests/test_data_models.py

@@ -1,58 +0,0 @@
-from django.db import connection
-from django.test import TestCase
-from django.test.utils import CaptureQueriesContext
-
-from documents.data_models import DocumentMetadataOverrides
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.tests.factories import DocumentFactory
-from documents.tests.utils import DirectoriesMixin
-
-
-class TestDocumentMetadataOverridesFromDocument(DirectoriesMixin, TestCase):
-    def test_from_document_batches_custom_field_lookup_after_refresh_from_db(
-        self,
-    ) -> None:
-        """
-        GIVEN:
-            - A document has several custom field values
-            - The document instance has just been refreshed from the database,
-              which drops any prefetched related objects (as
-              send_websocket_document_updated does before building overrides)
-        WHEN:
-            - DocumentMetadataOverrides.from_document() reads the document's
-              custom field values
-        THEN:
-            - The referenced CustomField objects are resolved with a single
-              query, not one query per custom field
-        """
-        doc = DocumentFactory(mime_type="application/pdf")
-        for i in range(5):
-            CustomFieldInstance.objects.create(
-                document=doc,
-                field=CustomField.objects.create(
-                    name=f"Test Custom Field {i}",
-                    data_type=CustomField.FieldDataType.STRING,
-                ),
-                value_text="value",
-            )
-
-        doc.refresh_from_db()
-
-        with CaptureQueriesContext(connection) as ctx:
-            overrides = DocumentMetadataOverrides.from_document(doc)
-
-        self.assertEqual(len(overrides.custom_fields), 5)
-        unbatched_field_lookups = [
-            query
-            for query in ctx.captured_queries
-            if 'FROM "documents_customfield" WHERE "documents_customfield"."id"'
-            in query["sql"]
-        ]
-        self.assertEqual(
-            unbatched_field_lookups,
-            [],
-            "Expected CustomField data to come from the CustomFieldInstance "
-            "join, not a separate per-instance lookup, "
-            f"got: {unbatched_field_lookups}",
-        )

src/documents/views.py

@@ -178,7 +178,7 @@ from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
-from documents.permissions import set_permissions_for_objects
+from documents.permissions import set_permissions_for_object
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
 from documents.search import SearchHit
@@ -2794,8 +2794,11 @@ class DocumentSelectionMixin:
             for key, value in filters.items()
             if key not in _TANTIVY_SEARCH_PARAM_NAMES
         }
+        # Operations are addressed to roots, a caller that wants
+        # to act on a specific version passes its id explicitly instead
         permitted_documents = Document.objects.filter(
             id__in=permitted_document_ids(user),
+            root_document__isnull=True,
         )
         # orm-filtered docs
         filtered_documents = DocumentFilterSet(
@@ -4917,12 +4920,12 @@ class BulkEditObjectsView(PassUserMixin):
                     qs_owner_update.update(owner=owner)
 
                 if "permissions" in serializer.validated_data:
-                    set_permissions_for_objects(
-                        permissions=permissions,
-                        model=object_class,
-                        pks=qs.values_list("pk", flat=True),
-                        merge=merge,
-                    )
+                    for obj in qs:
+                        set_permissions_for_object(
+                            permissions=permissions,
+                            object=obj,
+                            merge=merge,
+                        )
 
             except Exception as e:
                 logger.warning(

src/paperless_mail/tests/test_api.py

@@ -253,6 +253,81 @@ class TestAPIMailAccounts(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data["success"], True)
 
+    def test_mail_account_test_existing_no_global_perms(self) -> None:
+        """
+        GIVEN:
+            - Existing account without an owner
+            - User without any mail account permissions
+        WHEN:
+            - API call is made to test the account by id
+        THEN:
+            - API returns forbidden
+        """
+        account = MailAccountFactory(
+            username="admin",
+            password="secret",
+            imap_server="server.example.com",
+            imap_port=443,
+            owner=None,
+        )
+        user = User.objects.create_user(username="no_perms")
+        self.client.force_authenticate(user=user)
+
+        response = self.client.post(
+            f"{self.ENDPOINT}test/",
+            json.dumps(
+                {
+                    "id": account.pk,
+                    "imap_server": "server.example.com",
+                    "imap_port": 443,
+                    "imap_security": MailAccount.ImapSecurity.SSL,
+                    "username": "admin",
+                    "password": "******",
+                },
+            ),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(response.content.decode(), "Insufficient permissions")
+
+    def test_mail_account_test_existing_object_perms_only(self) -> None:
+        """
+        GIVEN:
+            - Existing account owned by another user
+            - User with an object level grant but no global change permission
+        WHEN:
+            - API call is made to test the account by id
+        THEN:
+            - API returns forbidden
+        """
+        owner = User.objects.create_user(username="account_owner")
+        account = MailAccountFactory(
+            username="admin",
+            password="secret",
+            imap_server="server.example.com",
+            imap_port=443,
+            owner=owner,
+        )
+        user = User.objects.create_user(username="object_perms_only")
+        assign_perm("change_mailaccount", user, account)
+        self.client.force_authenticate(user=user)
+
+        response = self.client.post(
+            f"{self.ENDPOINT}test/",
+            json.dumps(
+                {
+                    "id": account.pk,
+                    "imap_server": "server.example.com",
+                    "imap_port": 443,
+                    "imap_security": MailAccount.ImapSecurity.SSL,
+                    "username": "admin",
+                    "password": "******",
+                },
+            ),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_mail_account_test_existing_nonexistent_id_forbidden(self) -> None:
         response = self.client.post(
             f"{self.ENDPOINT}test/",

src/paperless_mail/tests/test_mail.py

@@ -2195,7 +2195,9 @@ class TestMailAccountTestView(APITestCase):
             password="testpassword",
         )
         self.user.user_permissions.add(
-            *Permission.objects.filter(codename__in=["add_mailaccount"]),
+            *Permission.objects.filter(
+                codename__in=["add_mailaccount", "change_mailaccount"],
+            ),
         )
         self.user.save()
         self.client.force_authenticate(user=self.user)

src/paperless_mail/views.py

@@ -106,7 +106,9 @@ class MailAccountViewSet(PassUserMixin, ModelViewSet[MailAccount]):
             except (TypeError, ValueError, MailAccount.DoesNotExist):
                 return HttpResponseForbidden("Insufficient permissions")
 
-            if not has_perms_owner_aware(
+            if not request.user.has_perms(
+                ["paperless_mail.change_mailaccount"],
+            ) or not has_perms_owner_aware(
                 request.user,
                 "change_mailaccount",
                 existing_account,

uv.lock

@@ -4,11 +4,11 @@ requires-python = ">=3.11"
 resolution-markers = [
     "python_full_version >= '3.15' and sys_platform == 'darwin'",
     "python_full_version >= '3.15' and sys_platform == 'linux'",
-    "python_full_version == '3.14.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
-    "python_full_version == '3.14.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
     "python_full_version >= '3.12' and python_full_version < '3.15' and sys_platform == 'darwin'",
     "python_full_version == '3.12.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
     "python_full_version == '3.12.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
+    "python_full_version == '3.14.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
+    "python_full_version == '3.14.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
     "(python_full_version >= '3.12' and python_full_version < '3.15' and platform_machine != 'aarch64' and platform_machine != 'x86_64' and sys_platform == 'linux') or (python_full_version == '3.13.*' and platform_machine == 'aarch64' and sys_platform == 'linux') or (python_full_version == '3.13.*' and platform_machine == 'x86_64' and sys_platform == 'linux')",
     "python_full_version < '3.12' and sys_platform == 'darwin'",
     "python_full_version < '3.12' and sys_platform == 'linux'",
@@ -1175,6 +1175,14 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/9d/76/d08f5c79f7643dbff4512605c28b75481966ed6c8cc9b397c9dd2ee91cd1/drf_spectacular_sidecar-2026.7.1-py3-none-any.whl", hash = "sha256:bc6d50c9b64660e45e09296d39553b3e759eedd825fc41d631ee5b3f88e0c5de", size = 2617384, upload-time = "2026-07-01T13:39:03.787Z" },
 ]
 
+[[package]]
+name = "drf-writable-nested"
+version = "0.7.2"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e8/57/df87d92fbfc3f0f2ef1a49c47f2a83389a4a13b7acf62b8bf7b223627d82/drf_writable_nested-0.7.2-py3-none-any.whl", hash = "sha256:4a3d2737c1cbfafa690e30236b169112e5b23cfe3d288f3992b0651a1b828c4d", size = 10570, upload-time = "2025-03-10T19:59:05.482Z" },
+]
+
 [[package]]
 name = "execnet"
 version = "2.1.2"
@@ -2887,6 +2895,7 @@ dependencies = [
     { name = "djangorestframework" },
     { name = "drf-spectacular" },
     { name = "drf-spectacular-sidecar" },
+    { name = "drf-writable-nested" },
     { name = "filelock" },
     { name = "flower" },
     { name = "gotenberg-client" },
@@ -3036,6 +3045,7 @@ requires-dist = [
     { name = "djangorestframework", specifier = "~=3.16" },
     { name = "drf-spectacular", specifier = "~=0.30" },
     { name = "drf-spectacular-sidecar", specifier = "~=2026.7.1" },
+    { name = "drf-writable-nested", specifier = "~=0.7.1" },
     { name = "filelock", specifier = "~=3.32.0" },
     { name = "flower", specifier = "~=2.0.1" },
     { name = "gotenberg-client", specifier = "~=0.14.0" },
@@ -5004,10 +5014,10 @@ version = "2.13.0+cpu"
 source = { registry = "https://download.pytorch.org/whl/cpu" }
 resolution-markers = [
     "python_full_version >= '3.15' and sys_platform == 'linux'",
-    "python_full_version == '3.14.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
-    "python_full_version == '3.14.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
     "python_full_version == '3.12.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
     "python_full_version == '3.12.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
+    "python_full_version == '3.14.*' and platform_machine == 'x86_64' and sys_platform == 'linux'",
+    "python_full_version == '3.14.*' and platform_machine == 'aarch64' and sys_platform == 'linux'",
     "(python_full_version >= '3.12' and python_full_version < '3.15' and platform_machine != 'aarch64' and platform_machine != 'x86_64' and sys_platform == 'linux') or (python_full_version == '3.13.*' and platform_machine == 'aarch64' and sys_platform == 'linux') or (python_full_version == '3.13.*' and platform_machine == 'x86_64' and sys_platform == 'linux')",
     "python_full_version < '3.12' and sys_platform == 'linux'",
 ]