from __future__ import annotations

import functools
import inspect
import json
import logging
import operator
from contextlib import contextmanager
from decimal import Decimal
from decimal import InvalidOperation
from typing import TYPE_CHECKING
from typing import Any

from django.contrib.contenttypes.models import ContentType
from django.db.models import Case
from django.db.models import CharField
from django.db.models import Count
from django.db.models import Exists
from django.db.models import IntegerField
from django.db.models import OuterRef
from django.db.models import Q
from django.db.models import Subquery
from django.db.models import Sum
from django.db.models import Value
from django.db.models import When
from django.db.models.functions import Cast
from django.db.models.functions import NullIf
from django.utils.translation import gettext_lazy as _
from django_filters import DateFilter
from django_filters.rest_framework import BooleanFilter
from django_filters.rest_framework import CharFilter
from django_filters.rest_framework import DateTimeFilter
from django_filters.rest_framework import Filter
from django_filters.rest_framework import FilterSet
from django_filters.rest_framework import MultipleChoiceFilter
from drf_spectacular.utils import extend_schema_field
from guardian.utils import get_group_obj_perms_model
from guardian.utils import get_user_obj_perms_model
from rest_framework import serializers
from rest_framework.filters import BaseFilterBackend
from rest_framework.filters import OrderingFilter

from documents.models import Correspondent
from documents.models import CustomField
from documents.models import CustomFieldInstance
from documents.models import Document
from documents.models import DocumentType
from documents.models import PaperlessTask
from documents.models import ShareLink
from documents.models import ShareLinkBundle
from documents.models import StoragePath
from documents.models import Tag
from documents.permissions import permitted_document_ids
from documents.permissions import permitted_object_ids
from documents.versioning import ensure_effective_content

if TYPE_CHECKING:
    from collections.abc import Callable

CHAR_KWARGS = ["istartswith", "iendswith", "icontains", "iexact"]
ID_KWARGS = ["in", "exact"]
INT_KWARGS = ["exact", "gt", "gte", "lt", "lte", "isnull"]
DATE_KWARGS = [
    "year",
    "month",
    "day",
    "gt",
    "gte",
    "lt",
    "lte",
]
DATETIME_KWARGS = [
    "year",
    "month",
    "day",
    "date__gt",
    "date__gte",
    "gt",
    "gte",
    "date__lt",
    "date__lte",
    "lt",
    "lte",
]

CUSTOM_FIELD_QUERY_MAX_DEPTH = 10
CUSTOM_FIELD_QUERY_MAX_ATOMS = 20

logger = logging.getLogger("paperless.api")


class CorrespondentFilterSet(FilterSet):
    class Meta:
        model = Correspondent
        fields = {
            "id": ID_KWARGS,
            "name": CHAR_KWARGS,
        }


class TagFilterSet(FilterSet):
    class Meta:
        model = Tag
        fields = {
            "id": ID_KWARGS,
            "name": CHAR_KWARGS,
        }

    is_root = BooleanFilter(
        label="Is root tag",
        field_name="tn_parent",
        lookup_expr="isnull",
    )


class DocumentTypeFilterSet(FilterSet):
    class Meta:
        model = DocumentType
        fields = {
            "id": ID_KWARGS,
            "name": CHAR_KWARGS,
        }


class StoragePathFilterSet(FilterSet):
    class Meta:
        model = StoragePath
        fields = {
            "id": ID_KWARGS,
            "name": CHAR_KWARGS,
            "path": CHAR_KWARGS,
        }


class ObjectFilter(Filter):
    def __init__(self, *, exclude=False, in_list=False, field_name="") -> None:
        super().__init__()
        self.exclude = exclude
        self.in_list = in_list
        self.field_name = field_name

    def filter(self, qs, value):
        if not value:
            return qs

        try:
            object_ids = [int(x) for x in value.split(",")]
        except ValueError:
            return qs

        if self.in_list:
            qs = qs.filter(**{f"{self.field_name}__id__in": object_ids}).distinct()
        else:
            for obj_id in object_ids:
                if self.exclude:
                    qs = qs.exclude(**{f"{self.field_name}__id": obj_id})
                else:
                    qs = qs.filter(**{f"{self.field_name}__id": obj_id})

        return qs


@extend_schema_field(serializers.BooleanField)
class InboxFilter(Filter):
    def filter(self, qs, value):
        if value == "true":
            # A document can have more than one tag flagged as an inbox tag
            # (nothing enforces uniqueness), so this join can multiply rows.
            return qs.filter(tags__is_inbox_tag=True).distinct()
        elif value == "false":
            return qs.exclude(tags__is_inbox_tag=True)
        else:
            return qs


@extend_schema_field(serializers.CharField)
class TitleContentFilter(Filter):
    # Deprecated but retained for existing saved views. UI uses Tantivy-backed `text` / `title_search` params.
    def filter(self, qs: Any, value: Any) -> Any:
        value = value.strip() if isinstance(value, str) else value
        if value:
            logger.warning(
                "Deprecated document filter parameter 'title_content' used; use `text` instead.",
            )
            return ensure_effective_content(qs).filter(
                Q(title__icontains=value) | Q(effective_content__icontains=value),
            )
        else:
            return qs


@extend_schema_field(serializers.CharField)
class EffectiveContentFilter(Filter):
    def filter(self, qs: Any, value: Any) -> Any:
        value = value.strip() if isinstance(value, str) else value
        if not value:
            return qs
        return ensure_effective_content(qs).filter(
            **{f"effective_content__{self.lookup_expr}": value},
        )


@extend_schema_field(serializers.BooleanField)
class SharedByUser(Filter):
    def filter(self, qs, value):
        ctype = ContentType.objects.get_for_model(self.model)
        UserObjectPermission = get_user_obj_perms_model()
        GroupObjectPermission = get_group_obj_perms_model()
        # see https://github.com/paperless-ngx/paperless-ngx/issues/5392, we limit subqueries
        # to 1 because Postgres doesn't like returning > 1 row, but all we care about is > 0
        return (
            qs.filter(
                owner_id=value,
            )
            .annotate(
                num_shared_users=Count(
                    UserObjectPermission.objects.filter(
                        content_type=ctype,
                        object_pk=Cast(OuterRef("pk"), CharField()),
                    ).values("user_id")[:1],
                ),
            )
            .annotate(
                num_shared_groups=Count(
                    GroupObjectPermission.objects.filter(
                        content_type=ctype,
                        object_pk=Cast(OuterRef("pk"), CharField()),
                    ).values("group_id")[:1],
                ),
            )
            .filter(
                Q(num_shared_users__gt=0) | Q(num_shared_groups__gt=0),
            )
            if value is not None
            else qs
        )


class CustomFieldFilterSet(FilterSet):
    class Meta:
        model = CustomField
        fields = {
            "id": ID_KWARGS,
            "name": CHAR_KWARGS,
        }


@extend_schema_field(serializers.CharField)
class CustomFieldsFilter(Filter):
    def filter(self, qs, value):
        value = value.strip() if isinstance(value, str) else value
        if value:
            logger.warning(
                "Deprecated document filter parameter 'custom_fields__icontains' used; use `custom_field_query` or advanced Tantivy field syntax instead.",
            )
            fields_with_matching_selects = CustomField.objects.filter(
                extra_data__icontains=value,
            )
            option_ids = []
            if fields_with_matching_selects.count() > 0:
                for field in fields_with_matching_selects:
                    options = field.extra_data.get("select_options", [])
                    for _, option in enumerate(options):
                        if option.get("label").lower().find(value.lower()) != -1:
                            option_ids.extend([option.get("id")])
            # A document with multiple custom field instances can match more
            # than one of these OR-ed branches (or the same branch via
            # different fields), each via its own join to custom_fields --
            # dedupe explicitly rather than relying on the caller to.
            return (
                qs.filter(custom_fields__field__name__icontains=value)
                | qs.filter(custom_fields__value_text__icontains=value)
                | qs.filter(custom_fields__value_bool__icontains=value)
                | qs.filter(custom_fields__value_int__icontains=value)
                | qs.filter(custom_fields__value_float__icontains=value)
                | qs.filter(custom_fields__value_date__icontains=value)
                | qs.filter(custom_fields__value_url__icontains=value)
                | qs.filter(custom_fields__value_monetary__icontains=value)
                | qs.filter(custom_fields__value_document_ids__icontains=value)
                | qs.filter(custom_fields__value_select__in=option_ids)
                | qs.filter(custom_fields__value_long_text__icontains=value)
            ).distinct()
        else:
            return qs


class MimeTypeFilter(Filter):
    def filter(self, qs, value):
        value = value.strip() if isinstance(value, str) else value
        if value:
            return qs.filter(mime_type__icontains=value)
        else:
            return qs


class MonetaryAmountField(serializers.Field):
    """
    Accepts either a plain decimal string ("100", "100.00") or a currency-prefixed
    string ("USD100.00") and returns the numeric amount as a Decimal.

    Mirrors the logic of the value_monetary_amount generated field: if the value
    starts with a non-digit, the first 3 characters are treated as a currency code
    (ISO 4217) and stripped before parsing. This preserves backwards compatibility
    with saved views that stored a currency-prefixed string as the filter value.
    """

    default_error_messages = {"invalid": "A valid number is required."}

    def to_internal_value(self, data):
        if not isinstance(data, str | int | float):
            self.fail("invalid")
        value = str(data).strip()
        if value and not value[0].isdigit() and value[0] != "-":
            value = value[3:]  # strip 3-char ISO 4217 currency code
        try:
            return Decimal(value)
        except InvalidOperation:
            self.fail("invalid")

    def to_representation(self, value):
        return str(value)


class SelectField(serializers.CharField):
    def __init__(self, custom_field: CustomField) -> None:
        self._options = custom_field.extra_data["select_options"]
        super().__init__(max_length=16)

    def to_internal_value(self, data):
        # If the supplied value is the option label instead of the ID
        try:
            data = next(
                option.get("id")
                for option in self._options
                if option.get("label") == data
            )
        except StopIteration:
            pass
        return super().to_internal_value(data)


def handle_validation_prefix(func: Callable):
    """
    Catch ValidationErrors raised by the wrapped function
    and add a prefix to the exception detail to track what causes the exception,
    similar to nested serializers.
    """

    def wrapper(*args, validation_prefix=None, **kwargs):
        try:
            return func(*args, **kwargs)
        except serializers.ValidationError as e:
            raise serializers.ValidationError({validation_prefix: e.detail})

    # Update the signature to include the validation_prefix argument
    old_sig = inspect.signature(func)
    new_param = inspect.Parameter("validation_prefix", inspect.Parameter.KEYWORD_ONLY)
    new_sig = old_sig.replace(parameters=[*old_sig.parameters.values(), new_param])

    # Apply functools.wraps and manually set the new signature
    functools.update_wrapper(wrapper, func)
    wrapper.__signature__ = new_sig

    return wrapper


class CustomFieldQueryParser:
    EXPR_BY_CATEGORY = {
        "basic": ["exact", "in", "isnull", "exists"],
        "string": [
            "icontains",
            "istartswith",
            "iendswith",
        ],
        "arithmetic": [
            "gt",
            "gte",
            "lt",
            "lte",
            "range",
        ],
        "containment": ["contains"],
    }

    SUPPORTED_EXPR_CATEGORIES = {
        CustomField.FieldDataType.STRING: ("basic", "string"),
        CustomField.FieldDataType.URL: ("basic", "string"),
        CustomField.FieldDataType.DATE: ("basic", "arithmetic"),
        CustomField.FieldDataType.BOOL: ("basic",),
        CustomField.FieldDataType.INT: ("basic", "arithmetic"),
        CustomField.FieldDataType.FLOAT: ("basic", "arithmetic"),
        CustomField.FieldDataType.MONETARY: ("basic", "string", "arithmetic"),
        CustomField.FieldDataType.DOCUMENTLINK: ("basic", "containment"),
        CustomField.FieldDataType.SELECT: ("basic",),
        CustomField.FieldDataType.LONG_TEXT: ("basic", "string"),
    }

    DATE_COMPONENTS = [
        "year",
        "iso_year",
        "month",
        "day",
        "week",
        "week_day",
        "iso_week_day",
        "quarter",
    ]

    def __init__(
        self,
        validation_prefix,
        max_query_depth=10,
        max_atom_count=20,
    ) -> None:
        """
        A helper class that parses the query string into a `django.db.models.Q` for filtering
        documents based on custom field values.

        The syntax of the query expression is illustrated with the below pseudo code rules:
        1. parse([`custom_field`, "exists", true]):
            matches documents with Q(custom_fields__field=`custom_field`)
        2. parse([`custom_field`, "exists", false]):
            matches documents with ~Q(custom_fields__field=`custom_field`)
        3. parse([`custom_field`, `op`, `value`]):
            matches documents with
            Q(custom_fields__field=`custom_field`, custom_fields__value_`type`__`op`= `value`)
        4. parse(["AND", [`q0`, `q1`, ..., `qn`]])
            -> parse(`q0`) & parse(`q1`) & ... & parse(`qn`)
        5. parse(["OR", [`q0`, `q1`, ..., `qn`]])
            -> parse(`q0`) | parse(`q1`) | ... | parse(`qn`)
        6. parse(["NOT", `q`])
            -> ~parse(`q`)

        Args:
            validation_prefix: Used to generate the ValidationError message.
            max_query_depth: Limits the maximum nesting depth of queries.
            max_atom_count: Limits the maximum number of atoms (i.e., rule 1, 2, 3) in the query.

        `max_query_depth` and `max_atom_count` can be set to guard against generating arbitrarily
        complex SQL queries.
        """
        self._custom_fields: dict[int | str, CustomField] = {}
        self._validation_prefix = validation_prefix
        # Dummy ModelSerializer used to convert a Django models.Field to serializers.Field.
        self._model_serializer = serializers.ModelSerializer()
        # Used for sanity check
        self._max_query_depth = max_query_depth
        self._max_atom_count = max_atom_count
        self._current_depth = 0
        self._atom_count = 0
        # The set of annotations that we need to apply to the queryset
        self._annotations = {}

    def parse(self, query: str) -> tuple[Q, dict[str, Count]]:
        """
        Parses the query string into a `django.db.models.Q`
        and a set of annotations to be applied to the queryset.
        """
        try:
            expr = json.loads(query)
        except json.JSONDecodeError:
            raise serializers.ValidationError(
                {self._validation_prefix: [_("Value must be valid JSON.")]},
            )
        return (
            self._parse_expr(expr, validation_prefix=self._validation_prefix),
            self._annotations,
        )

    @handle_validation_prefix
    def _parse_expr(self, expr) -> Q:
        """
        Applies rule (1, 2, 3) or (4, 5, 6) based on the length of the expr.
        """
        with self._track_query_depth():
            if isinstance(expr, list | tuple):
                if len(expr) == 2:
                    return self._parse_logical_expr(*expr)
                elif len(expr) == 3:
                    return self._parse_atom(*expr)
            raise serializers.ValidationError(
                [_("Invalid custom field query expression")],
            )

    @handle_validation_prefix
    def _parse_expr_list(self, exprs) -> list[Q]:
        """
        Handles [`q0`, `q1`, ..., `qn`] in rule 4 & 5.
        """
        if not isinstance(exprs, list | tuple) or not exprs:
            raise serializers.ValidationError(
                [_("Invalid expression list. Must be nonempty.")],
            )
        return [
            self._parse_expr(expr, validation_prefix=i) for i, expr in enumerate(exprs)
        ]

    def _parse_logical_expr(self, op, args) -> Q:
        """
        Handles rule 4, 5, 6.
        """
        op_lower = op.lower()

        if op_lower == "not":
            return ~self._parse_expr(args, validation_prefix=1)

        if op_lower == "and":
            op_func = operator.and_
        elif op_lower == "or":
            op_func = operator.or_
        else:
            raise serializers.ValidationError(
                {"0": [_("Invalid logical operator {op!r}").format(op=op)]},
            )

        qs = self._parse_expr_list(args, validation_prefix="1")
        return functools.reduce(op_func, qs)

    def _parse_atom(self, id_or_name, op, value) -> Q:
        """
        Handles rule 1, 2, 3.
        """
        # Guard against queries with too many conditions.
        self._atom_count += 1
        if self._atom_count > self._max_atom_count:
            raise serializers.ValidationError(
                [_("Maximum number of query conditions exceeded.")],
            )

        custom_field = self._get_custom_field(id_or_name, validation_prefix="0")
        op = self._validate_atom_op(custom_field, op, validation_prefix="1")
        value = self._validate_atom_value(
            custom_field,
            op,
            value,
            validation_prefix="2",
        )

        # Needed because not all DB backends support Array __contains
        if (
            custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
            and op == "contains"
        ):
            return self._parse_atom_doc_link_contains(custom_field, value)

        value_field_name = CustomFieldInstance.get_value_field_name(
            custom_field.data_type,
        )
        if custom_field.data_type == CustomField.FieldDataType.MONETARY and (
            op in self.EXPR_BY_CATEGORY["arithmetic"] or op in {"exact", "in"}
        ):
            value_field_name = "value_monetary_amount"
        has_field = Q(custom_fields__field=custom_field)

        # We need to use an annotation here because different atoms
        # might be referring to different instances of custom fields.
        annotation_name = f"_custom_field_filter_{len(self._annotations)}"

        # Our special exists operator.
        if op == "exists":
            annotation = Count("custom_fields", filter=has_field)
            # A Document should have > 0 match if it has this field, or 0 if doesn't.
            query_op = "gt" if value else "exact"
            query = Q(**{f"{annotation_name}__{query_op}": 0})
        else:
            # Check if 1) custom field name matches, and 2) value satisfies condition
            field_filter = has_field & Q(
                **{f"custom_fields__{value_field_name}__{op}": value},
            )
            # Annotate how many matching custom fields each document has
            annotation = Count("custom_fields", filter=field_filter)
            # Filter document by count
            query = Q(**{f"{annotation_name}__gt": 0})

        self._annotations[annotation_name] = annotation
        return query

    @handle_validation_prefix
    def _get_custom_field(self, id_or_name):
        """Get the CustomField instance by id or name."""
        if id_or_name in self._custom_fields:
            return self._custom_fields[id_or_name]

        kwargs = (
            {"id": id_or_name} if isinstance(id_or_name, int) else {"name": id_or_name}
        )
        try:
            custom_field = CustomField.objects.get(**kwargs)
        except CustomField.DoesNotExist:
            raise serializers.ValidationError(
                [_("{name!r} is not a valid custom field.").format(name=id_or_name)],
            )
        self._custom_fields[custom_field.id] = custom_field
        self._custom_fields[custom_field.name] = custom_field
        return custom_field

    @staticmethod
    def _split_op(full_op):
        *prefix, op = str(full_op).rsplit("__", maxsplit=1)
        prefix = prefix[0] if prefix else None
        return prefix, op

    @handle_validation_prefix
    def _validate_atom_op(self, custom_field, raw_op):
        """Check if the `op` is compatible with the type of the custom field."""
        prefix, op = self._split_op(raw_op)

        # Check if the operator is supported for the current data_type.
        supported = False
        for category in self.SUPPORTED_EXPR_CATEGORIES[custom_field.data_type]:
            if op in self.EXPR_BY_CATEGORY[category]:
                supported = True
                break

        # Check prefix
        if prefix is not None:
            if (
                prefix in self.DATE_COMPONENTS
                and custom_field.data_type == CustomField.FieldDataType.DATE
            ):
                pass  # ok - e.g., "year__exact" for date field
            else:
                supported = False  # anything else is invalid

        if not supported:
            raise serializers.ValidationError(
                [
                    _("{data_type} does not support query expr {expr!r}.").format(
                        data_type=custom_field.data_type,
                        expr=raw_op,
                    ),
                ],
            )

        return raw_op

    def _get_serializer_field(self, custom_field, full_op):
        """Return a serializers.Field for value validation."""
        prefix, op = self._split_op(full_op)
        field = None

        if op in ("isnull", "exists"):
            # `isnull` takes either True or False regardless of the data_type.
            field = serializers.BooleanField()
        elif (
            custom_field.data_type == CustomField.FieldDataType.DATE
            and prefix in self.DATE_COMPONENTS
        ):
            # DateField admits queries in the form of `year__exact`, etc. These take integers.
            field = serializers.IntegerField()
        elif custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
            # We can be more specific here and make sure the value is a list.
            field = serializers.ListField(child=serializers.IntegerField())
        elif custom_field.data_type == CustomField.FieldDataType.SELECT:
            # We use this custom field to permit SELECT option names.
            field = SelectField(custom_field)
        elif custom_field.data_type == CustomField.FieldDataType.URL:
            # For URL fields we don't need to be strict about validation (e.g., for istartswith).
            field = serializers.CharField()
        elif custom_field.data_type == CustomField.FieldDataType.MONETARY and (
            op in self.EXPR_BY_CATEGORY["arithmetic"] or op in {"exact", "in"}
        ):
            # These ops compare against value_monetary_amount (a DecimalField).
            # MonetaryAmountField accepts both "100" and "USD100.00" for backwards
            # compatibility with saved views that stored currency-prefixed values.
            field = MonetaryAmountField()
        else:
            # The general case: inferred from the corresponding field in CustomFieldInstance.
            value_field_name = CustomFieldInstance.get_value_field_name(
                custom_field.data_type,
            )
            model_field = CustomFieldInstance._meta.get_field(value_field_name)
            field_name = model_field.deconstruct()[0]
            field_class, field_kwargs = self._model_serializer.build_standard_field(
                field_name,
                model_field,
            )
            field = field_class(**field_kwargs)
            field.allow_null = False

            # Need to set allow_blank manually because of the inconsistency in CustomFieldInstance validation.
            # See https://github.com/paperless-ngx/paperless-ngx/issues/7361.
            if isinstance(field, serializers.CharField):
                field.allow_blank = True

        if op == "in":
            # `in` takes a list of values.
            field = serializers.ListField(child=field, allow_empty=False)
        elif op == "range":
            # `range` takes a list of values, i.e., [start, end].
            field = serializers.ListField(
                child=field,
                min_length=2,
                max_length=2,
            )

        return field

    @handle_validation_prefix
    def _validate_atom_value(self, custom_field, op, value):
        """Check if `value` is valid for the custom field and `op`. Returns the validated value."""
        serializer_field = self._get_serializer_field(custom_field, op)
        return serializer_field.run_validation(value)

    def _parse_atom_doc_link_contains(self, custom_field, value) -> Q:
        """
        Handles document link `contains` in a way that is supported by all DB backends.
        """

        # If the value is an empty set,
        # this is trivially true for any document with not null document links.
        if not value:
            return Q(
                custom_fields__field=custom_field,
                custom_fields__value_document_ids__isnull=False,
            )

        # First we look up reverse links from the requested documents.
        # Scoped to this specific field (not just any document link field) and
        # excluding unset instances, which have a null value_document_ids and
        # are equivalent to having no reverse link at all.
        links = CustomFieldInstance.objects.filter(
            document_id__in=value,
            field=custom_field,
            value_document_ids__isnull=False,
        )

        # Check if any of the requested IDs are missing.
        missing_ids = set(value) - set(link.document_id for link in links)
        if missing_ids:
            # The result should be an empty set in this case.
            return Q(id__in=[])

        # Take the intersection of the reverse links - this should be what we are looking for.
        document_ids_we_want = functools.reduce(
            operator.and_,
            (set(link.value_document_ids) for link in links),
        )

        return Q(id__in=document_ids_we_want)

    @contextmanager
    def _track_query_depth(self):
        # guard against queries that are too deeply nested
        self._current_depth += 1
        if self._current_depth > self._max_query_depth:
            raise serializers.ValidationError([_("Maximum nesting depth exceeded.")])
        try:
            yield
        finally:
            self._current_depth -= 1


@extend_schema_field(serializers.CharField)
class CustomFieldQueryFilter(Filter):
    def __init__(self, validation_prefix) -> None:
        """
        A filter that filters documents based on custom field name and value.

        Args:
            validation_prefix: Used to generate the ValidationError message.
        """
        super().__init__()
        self._validation_prefix = validation_prefix

    def filter(self, qs, value):
        if not value:
            return qs

        parser = CustomFieldQueryParser(
            self._validation_prefix,
            max_query_depth=CUSTOM_FIELD_QUERY_MAX_DEPTH,
            max_atom_count=CUSTOM_FIELD_QUERY_MAX_ATOMS,
        )
        q, annotations = parser.parse(value)

        # The Count(...) annotations above require a GROUP BY/HAVING to evaluate.
        # Applying them directly to `qs` mixes that HAVING with `qs`'s existing
        # joins (e.g. repeated tag joins from tags__id__all, the object-permission
        # OR-filter), which some backends (e.g. MariaDB) fail to plan correctly,
        # raising "Unknown column ... in 'HAVING'". Evaluating the annotation on
        # an isolated queryset keeps the GROUP BY/HAVING self-contained.
        matching_ids = Document.objects.annotate(**annotations).filter(q).values("pk")
        return qs.filter(pk__in=matching_ids)


class DocumentFilterSet(FilterSet):
    has_duplicates = BooleanFilter(method="filter_has_duplicates")

    def __init__(self, *args: Any, user: Any = None, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._user = user

    is_tagged = BooleanFilter(
        label="Is tagged",
        field_name="tags",
        lookup_expr="isnull",
        exclude=True,
    )

    tags__id__all = ObjectFilter(field_name="tags")

    tags__id__none = ObjectFilter(field_name="tags", exclude=True)

    tags__id__in = ObjectFilter(field_name="tags", in_list=True)

    correspondent__id__none = ObjectFilter(field_name="correspondent", exclude=True)

    document_type__id__none = ObjectFilter(field_name="document_type", exclude=True)

    storage_path__id__none = ObjectFilter(field_name="storage_path", exclude=True)

    is_in_inbox = InboxFilter()

    # Deprecated, but keep for now for existing saved views
    title_content = TitleContentFilter()

    content__istartswith = EffectiveContentFilter(lookup_expr="istartswith")
    content__iendswith = EffectiveContentFilter(lookup_expr="iendswith")
    content__icontains = EffectiveContentFilter(lookup_expr="icontains")
    content__iexact = EffectiveContentFilter(lookup_expr="iexact")

    owner__id__none = ObjectFilter(field_name="owner", exclude=True)

    # Deprecated, UI no longer includes CF text-search mode, but keep for now for existing saved views
    custom_fields__icontains = CustomFieldsFilter()

    custom_fields__id__all = ObjectFilter(field_name="custom_fields__field")

    custom_fields__id__none = ObjectFilter(
        field_name="custom_fields__field",
        exclude=True,
    )

    custom_fields__id__in = ObjectFilter(
        field_name="custom_fields__field",
        in_list=True,
    )

    has_custom_fields = BooleanFilter(
        label="Has custom field",
        field_name="custom_fields",
        lookup_expr="isnull",
        exclude=True,
    )

    custom_field_query = CustomFieldQueryFilter("custom_field_query")

    shared_by__id = SharedByUser()

    mime_type = MimeTypeFilter()

    def filter_has_duplicates(self, queryset, name, value):
        if value is None:
            return queryset

        user = (
            self._user
            if self._user is not None
            else getattr(self.request, "user", None)
        )
        queryset = queryset.alias(
            nonempty_archive_checksum=NullIf("archive_checksum", Value("")),
        )

        visible_root_documents = Document.global_objects.filter(
            root_document__isnull=True,
            pk__in=permitted_document_ids(
                user,
                include_deleted=True,
            ),
        ).exclude(pk=OuterRef("pk"))
        # see serialisers._get_viewable_duplicates().
        matching_duplicates = visible_root_documents.filter(
            Q(checksum=OuterRef("checksum"))
            | Q(checksum=OuterRef("nonempty_archive_checksum"))
            | Q(archive_checksum=OuterRef("checksum"))
            | Q(archive_checksum=OuterRef("nonempty_archive_checksum")),
        )

        return queryset.alias(
            has_visible_duplicates=Exists(matching_duplicates),
        ).filter(has_visible_duplicates=value)

    # Backwards compatibility
    created__date__gt = DateFilter(field_name="created", lookup_expr="gt")
    created__date__gte = DateFilter(field_name="created", lookup_expr="gte")
    created__date__lt = DateFilter(field_name="created", lookup_expr="lt")
    created__date__lte = DateFilter(field_name="created", lookup_expr="lte")

    class Meta:
        model = Document
        fields = {
            "id": ID_KWARGS,
            "title": CHAR_KWARGS,
            "archive_serial_number": INT_KWARGS,
            "created": DATE_KWARGS,
            "added": DATETIME_KWARGS,
            "modified": DATETIME_KWARGS,
            "original_filename": CHAR_KWARGS,
            "checksum": CHAR_KWARGS,
            "correspondent": ["isnull"],
            "correspondent__id": ID_KWARGS,
            "correspondent__name": CHAR_KWARGS,
            "tags__id": ID_KWARGS,
            "tags__name": CHAR_KWARGS,
            "document_type": ["isnull"],
            "document_type__id": ID_KWARGS,
            "document_type__name": CHAR_KWARGS,
            "storage_path": ["isnull"],
            "storage_path__id": ID_KWARGS,
            "storage_path__name": CHAR_KWARGS,
            "owner": ["isnull"],
            "owner__id": ID_KWARGS,
            "custom_fields": ["icontains"],
        }


class ShareLinkFilterSet(FilterSet):
    class Meta:
        model = ShareLink
        fields = {
            "created": DATETIME_KWARGS,
            "expiration": DATETIME_KWARGS,
        }


class ShareLinkBundleFilterSet(FilterSet):
    documents = Filter(method="filter_documents")

    class Meta:
        model = ShareLinkBundle
        fields = {
            "created": DATETIME_KWARGS,
            "expiration": DATETIME_KWARGS,
            "status": ["exact"],
        }

    def filter_documents(self, queryset, name, value):
        ids = []
        if value:
            try:
                ids = [int(item) for item in value.split(",") if item]
            except ValueError:
                return queryset.none()
        if not ids:
            return queryset
        return queryset.filter(documents__in=ids).distinct()


class PaperlessTaskFilterSet(FilterSet):
    name = CharFilter(
        method="filter_name",
        label="Name",
    )

    result = CharFilter(
        method="filter_result",
        label="Result",
    )

    task_type = MultipleChoiceFilter(
        choices=PaperlessTask.TaskType.choices,
        label="Task Type",
    )

    trigger_source = MultipleChoiceFilter(
        choices=PaperlessTask.TriggerSource.choices,
        label="Trigger Source",
    )

    status = MultipleChoiceFilter(
        choices=PaperlessTask.Status.choices,
        label="Status",
    )

    is_complete = BooleanFilter(
        method="filter_is_complete",
        label="Is Complete",
    )

    acknowledged = BooleanFilter(
        label="Acknowledged",
        field_name="acknowledged",
    )

    date_created_after = DateTimeFilter(
        field_name="date_created",
        lookup_expr="gte",
        label="Created After",
    )

    date_created_before = DateTimeFilter(
        field_name="date_created",
        lookup_expr="lte",
        label="Created Before",
    )

    class Meta:
        model = PaperlessTask
        fields = [
            "task_type",
            "trigger_source",
            "status",
            "acknowledged",
            "owner",
            "name",
            "result",
        ]

    def filter_name(self, queryset, name, value):
        if not value:
            return queryset

        matching_task_types = [
            task_type
            for task_type, label in PaperlessTask.TaskType.choices
            if value.lower() in str(label).lower()
        ]
        matching_trigger_sources = [
            trigger_source
            for trigger_source, label in PaperlessTask.TriggerSource.choices
            if value.lower() in str(label).lower()
        ]

        return queryset.filter(
            Q(input_data__filename__icontains=value)
            | Q(task_type__in=matching_task_types)
            | Q(trigger_source__in=matching_trigger_sources),
        )

    def filter_result(self, queryset, name, value):
        if not value:
            return queryset

        query = Q(result_data__reason__icontains=value) | Q(
            result_data__error_message__icontains=value,
        )

        try:
            numeric_value = int(value)
        except (TypeError, ValueError):
            pass
        else:
            query |= Q(result_data__document_id=numeric_value) | Q(
                result_data__duplicate_of=numeric_value,
            )

        if "duplicate" in value.lower():
            query |= Q(result_data__duplicate_of__isnull=False)

        return queryset.filter(query)

    def filter_is_complete(self, queryset, name, value):
        if value:
            return queryset.filter(status__in=PaperlessTask.COMPLETE_STATUSES)
        return queryset.exclude(status__in=PaperlessTask.COMPLETE_STATUSES)


class PermittedObjectsFilter(BaseFilterBackend):
    """
    Filters a queryset down to objects the requesting user owns, are
    unowned, or (when ``include_granted`` is True) has an explicit
    user/group guardian permission on. Backed by ``permitted_object_ids``
    -- a single ``id__in`` subquery, not a join -- so it can't produce
    duplicate rows even when the base queryset already carries independent
    joins (e.g. multi-value ``tags__id__all`` filtering), and stays
    index-friendly at scale instead of falling back to guardian's
    varchar-cast join.

    Set ``include_granted = False`` on a subclass for endpoints that
    intentionally only show owned/unowned objects regardless of explicit
    shares (e.g. ``TrashView``).
    """

    include_granted: bool = True
    perm_codename: str | None = None

    def filter_queryset(self, request, queryset, view):
        # Before the superuser and owner-only paths, neither of which consults
        # permitted_object_ids. Scoped to authenticated users so anonymous
        # access (AnonymousUser.is_active is False) keeps its existing
        # unowned-only behaviour.
        if request.user.is_authenticated and not request.user.is_active:
            return queryset.none()
        if request.user.is_superuser:
            return queryset
        if not self.include_granted:
            return queryset.filter(Q(owner=request.user) | Q(owner__isnull=True))
        model = queryset.model
        perm = self.perm_codename or f"view_{model._meta.model_name}"
        return queryset.filter(
            id__in=permitted_object_ids(request.user, model, perm),
        )


class DocumentsOrderingFilter(OrderingFilter):
    field_name = "ordering"
    prefix = "custom_field_"

    def filter_queryset(self, request, queryset, view):
        param = request.query_params.get("ordering")
        if param and self.prefix in param:
            custom_field_id = int(param.split(self.prefix)[1])
            try:
                field = CustomField.objects.get(pk=custom_field_id)
            except CustomField.DoesNotExist:
                raise serializers.ValidationError(
                    {self.prefix + str(custom_field_id): [_("Custom field not found")]},
                )

            annotation = None
            match field.data_type:
                case (
                    CustomField.FieldDataType.STRING
                    | CustomField.FieldDataType.LONG_TEXT
                ):
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_text")[:1],
                    )
                case CustomField.FieldDataType.INT:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_int")[:1],
                    )
                case CustomField.FieldDataType.FLOAT:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_float")[:1],
                    )
                case CustomField.FieldDataType.DATE:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_date")[:1],
                    )
                case CustomField.FieldDataType.MONETARY:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_monetary_amount")[:1],
                    )
                case CustomField.FieldDataType.SELECT:
                    # Select options are a little more complicated since the value is the id of the option, not
                    # the label. Additionally, to support sqlite we can't use StringAgg, so we need to create a
                    # case statement for each option, setting the value to the index of the option in a list
                    # sorted by label, and then summing the results to give a single value for the annotation

                    select_options = sorted(
                        field.extra_data.get("select_options", []),
                        key=lambda x: x.get("label"),
                    )
                    whens = [
                        When(
                            custom_fields__field_id=custom_field_id,
                            custom_fields__value_select=option.get("id"),
                            then=Value(idx, output_field=IntegerField()),
                        )
                        for idx, option in enumerate(select_options)
                    ]
                    whens.append(
                        When(
                            custom_fields__field_id=custom_field_id,
                            custom_fields__value_select__isnull=True,
                            then=Value(
                                len(select_options),
                                output_field=IntegerField(),
                            ),
                        ),
                    )
                    annotation = Sum(
                        Case(
                            *whens,
                            default=Value(0),
                            output_field=IntegerField(),
                        ),
                    )
                case CustomField.FieldDataType.DOCUMENTLINK:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_document_ids")[:1],
                    )
                case CustomField.FieldDataType.URL:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_url")[:1],
                    )
                case CustomField.FieldDataType.BOOL:
                    annotation = Subquery(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ).values("value_bool")[:1],
                    )

            if not annotation:
                # Only happens if a new data type is added and not handled here
                raise ValueError("Invalid custom field data type")

            queryset = (
                queryset.annotate(
                    # We need to annotate the queryset with the custom field value
                    custom_field_value=annotation,
                    # We also need to annotate the queryset with a boolean for sorting whether the field exists
                    has_field=Exists(
                        CustomFieldInstance.objects.filter(
                            document_id=OuterRef("id"),
                            field_id=custom_field_id,
                        ),
                    ),
                )
                .order_by(
                    "-has_field",
                    param.replace(
                        self.prefix + str(custom_field_id),
                        "custom_field_value",
                    ),
                )
                .distinct()
            )

        return super().filter_queryset(request, queryset, view)
