import re
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Generic, Literal, TypeAlias, TypedDict, TypeVar

from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import ValidationError
from django.db import connection
from django.utils.translation import gettext as _
from pydantic import BaseModel, model_validator
from sqlalchemy.dialects import postgresql
from sqlalchemy.engine import Connection, Row
from sqlalchemy.sql import (
    ClauseElement,
    ColumnElement,
    Select,
    and_,
    column,
    false,
    func,
    literal,
    literal_column,
    not_,
    or_,
    select,
    table,
    true,
    union_all,
)
from sqlalchemy.sql.selectable import SelectBase
from sqlalchemy.types import ARRAY, Boolean, Integer, Text
from typing_extensions import override

from zerver.lib.addressee import get_user_profiles, get_user_profiles_by_ids
from zerver.lib.exceptions import ErrorCode, JsonableError, MissingAuthenticationError
from zerver.lib.message import (
    access_message,
    access_web_public_message,
    get_first_visible_message_id,
)
from zerver.lib.narrow_predicate import channel_operators, channels_operators
from zerver.lib.recipient_users import recipient_for_user_profiles
from zerver.lib.sqlalchemy_utils import get_sqlalchemy_connection
from zerver.lib.streams import (
    can_access_stream_history_by_id,
    can_access_stream_history_by_name,
    get_public_streams_queryset,
    get_stream_by_narrow_operand_access_unchecked,
    get_web_public_streams_queryset,
)
from zerver.lib.topic import maybe_rename_general_chat_to_empty_topic
from zerver.lib.topic_sqlalchemy import (
    get_followed_topic_condition_sa,
    get_resolved_topic_condition_sa,
    topic_column_sa,
    topic_match_sa,
)
from zerver.lib.types import Validator
from zerver.lib.user_groups import get_recursive_membership_groups
from zerver.lib.user_topics import exclude_stream_and_topic_mutes
from zerver.lib.validator import (
    check_bool,
    check_iso_datetime,
    check_required_string,
    check_string,
    check_string_or_int,
    check_string_or_int_list,
)
from zerver.models import (
    DirectMessageGroup,
    Message,
    Realm,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
)
from zerver.models.recipients import get_direct_message_group_user_ids
from zerver.models.users import (
    get_user_by_id_in_realm_including_cross_realm,
    get_user_including_cross_realm,
)


class NarrowParameter(BaseModel):
    operator: str
    operand: Any
    negated: bool = False

    @model_validator(mode="before")
    @classmethod
    def convert_term(cls, elem: dict[str, Any] | list[str]) -> dict[str, Any]:
        # We have to support a legacy tuple format.
        if isinstance(elem, list):
            if len(elem) != 2 or any(not isinstance(x, str) for x in elem):
                raise ValueError("element is not a string pair")
            return dict(operator=elem[0], operand=elem[1])

        elif isinstance(elem, dict):
            if "operand" not in elem or elem["operand"] is None:
                raise ValueError("operand is missing")

            if "operator" not in elem or elem["operator"] is None:
                raise ValueError("operator is missing")
            return elem
        else:
            raise ValueError("dict or list required")

    @model_validator(mode="after")
    def validate_terms(self) -> "NarrowParameter":
        # Make sure to sync this list to frontend also when adding a new operator that
        # supports integer IDs. Relevant code is located in web/src/message_fetch.ts
        # in handle_operators_supporting_id_based_api function where you will need to
        # update operators_supporting_id, or operators_supporting_ids array.
        operators_supporting_id = [
            *channel_operators,
            "id",
            "sender",
            "group-pm-with",
            "dm-including",
            "with",
        ]
        operators_supporting_ids = ["pm-with", "dm"]
        operators_non_empty_operand = {"search"}

        operator = self.operator
        if operator in operators_supporting_id:
            operand_validator: Validator[object] = check_string_or_int
        elif operator in operators_supporting_ids:
            operand_validator = check_string_or_int_list
        elif operator in operators_non_empty_operand:
            operand_validator = check_required_string
        else:
            operand_validator = check_string

        try:
            self.operand = operand_validator("operand", self.operand)
            self.operator = check_string("operator", self.operator)
            if self.negated is not None:
                self.negated = check_bool("negated", self.negated)
        except ValidationError as error:
            raise JsonableError(error.message)

        # whitelist the fields we care about for now
        return self


def is_spectator_compatible(narrow: Iterable[NarrowParameter]) -> bool:
    # This implementation should agree with is_spectator_compatible in hash_parser.ts.
    supported_operators = [
        *channel_operators,
        *channels_operators,
        "topic",
        "sender",
        "has",
        "search",
        "near",
        "id",
        "with",
    ]
    for element in narrow:
        operator = element.operator
        operand = element.operand

        if operator == "is" and operand == "resolved":
            continue
        if operator not in supported_operators:
            return False
    return True


def is_web_public_narrow(narrow: Iterable[NarrowParameter] | None) -> bool:
    if narrow is None:
        return False

    return any(
        # Web-public queries are only allowed for limited types of narrows.
        # term == {'operator': 'channels', 'operand': 'web-public', 'negated': False}
        # or term == {'operator': 'streams', 'operand': 'web-public', 'negated': False}
        term.operator in channels_operators
        and term.operand == "web-public"
        and term.negated is False
        for term in narrow
    )


LARGER_THAN_MAX_MESSAGE_ID = 10000000000000000


class AnchorInfo(TypedDict):
    type: Literal["message_id", "first_unread", "date"]
    value: int | datetime | None


DEFAULT_ANCHOR_INFO: AnchorInfo = AnchorInfo(type="first_unread", value=None)


class BadNarrowOperatorError(JsonableError):
    code = ErrorCode.BAD_NARROW
    data_fields = ["desc"]

    def __init__(self, desc: str) -> None:
        self.desc: str = desc

    @staticmethod
    @override
    def msg_format() -> str:
        return _("Invalid narrow operator: {desc}")


class InvalidOperatorCombinationError(JsonableError):
    code = ErrorCode.BAD_NARROW
    data_fields = ["desc"]

    def __init__(self, desc: str) -> None:
        self.desc: str = desc

    @staticmethod
    @override
    def msg_format() -> str:
        return _("Invalid narrow operator combination: {desc}")


ConditionTransform: TypeAlias = Callable[[ClauseElement], ClauseElement]

# These delimiters will not appear in rendered messages or HTML-escaped topics.
TS_START = "<ts-match>"
TS_STOP = "</ts-match>"


def ts_locs_array(
    config: ColumnElement[Text],
    text: ColumnElement[Text],
    tsquery: ColumnElement[Any],
) -> ColumnElement[ARRAY[Integer]]:
    options = f"HighlightAll = TRUE, StartSel = {TS_START}, StopSel = {TS_STOP}"
    delimited = func.ts_headline(config, text, tsquery, options, type_=Text)
    part = func.unnest(
        func.string_to_array(delimited, TS_START, type_=ARRAY(Text)), type_=Text
    ).column_valued()
    part_len = func.length(part, type_=Integer) - len(TS_STOP)
    match_pos = func.sum(part_len, type_=Integer).over(rows=(None, -1)) + len(TS_STOP)
    match_len = func.strpos(part, TS_STOP, type_=Integer) - 1
    return func.array(
        select(postgresql.array([match_pos, match_len])).offset(1).scalar_subquery(),
        type_=ARRAY(Integer),
    )


class NarrowBuilder:
    """
    Build up a SQLAlchemy query to find messages matching a narrow.
    """

    # This class has an important security invariant:
    #
    #   None of these methods ever *add* messages to a query's result.
    #
    # That is, the `add_term` method, and its helpers the `by_*` methods,
    # are passed a Select object representing a query for messages; they may
    # call some methods on it, and then they return a resulting Select
    # object.  Things these methods may do to the queries they handle
    # include
    #  * add conditions to filter out rows (i.e., messages), with `query.where`
    #  * add columns for more information on the same message, with `query.column`
    #  * add a join for more information on the same message
    #
    # Things they may not do include
    #  * anything that would pull in additional rows, or information on
    #    other messages.

    def __init__(
        self,
        user_profile: UserProfile | None,
        msg_id_column: ColumnElement[Integer],
        realm: Realm,
        is_web_public_query: bool = False,
    ) -> None:
        self.user_profile = user_profile
        self.msg_id_column = msg_id_column
        self.realm = realm
        self.is_web_public_query = is_web_public_query
        self.by_method_map = {
            "has": self.by_has,
            "in": self.by_in,
            "is": self.by_is,
            "channel": self.by_channel,
            # "stream" is a legacy alias for "channel"
            "stream": self.by_channel,
            "channels": self.by_channels,
            # "streams" is a legacy alias for "channels"
            "streams": self.by_channels,
            "topic": self.by_topic,
            "sender": self.by_sender,
            "near": self.by_near,
            "id": self.by_id,
            "search": self.by_search,
            "dm": self.by_dm,
            # "pm-with:" is a legacy alias for "dm:"
            "pm-with": self.by_dm,
            "dm-including": self.by_dm_including,
            # "group-pm-with:" was deprecated by the addition of "dm-including:"
            "group-pm-with": self.by_group_pm_with,
            # TODO/compatibility: Prior to commit a9b3a9c, the server implementation
            # for documented search operators with dashes, also implicitly supported
            # clients sending those same operators with underscores. We can remove
            # support for the below operators when support for the associated dashed
            # operator is removed.
            "pm_with": self.by_dm,
            "group_pm_with": self.by_group_pm_with,
        }
        self.is_channel_narrow = False
        self.is_dm_narrow = False

    def check_not_both_channel_and_dm_narrow(
        self,
        maybe_negate: ConditionTransform,
        is_dm_narrow: bool = False,
        is_channel_narrow: bool = False,
    ) -> None:
        if maybe_negate is not_:
            return
        if is_dm_narrow:
            self.is_dm_narrow = True
        if is_channel_narrow:
            self.is_channel_narrow = True
        if self.is_channel_narrow and self.is_dm_narrow:
            raise BadNarrowOperatorError(
                "No message can be both a channel message and direct message"
            )

    def add_term(self, query: Select, term: NarrowParameter) -> Select:
        """
        Extend the given query to one narrowed by the given term, and return the result.

        This method satisfies an important security property: the returned
        query never includes a message that the given query didn't.  In
        particular, if the given query will only find messages that a given
        user can legitimately see, then so will the returned query.
        """
        # To maintain the security property, we hold all the `by_*`
        # methods to the same criterion.  See the class's block comment
        # for details.

        operator = term.operator
        operand = term.operand

        negated = term.negated

        if operator in self.by_method_map:
            method = self.by_method_map[operator]
        else:
            raise BadNarrowOperatorError("unknown operator " + operator)

        if negated:
            maybe_negate: ConditionTransform = not_
        else:
            maybe_negate = lambda cond: cond

        return method(query, operand, maybe_negate)

    def by_has(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        if operand not in ["attachment", "image", "link", "reaction"]:
            raise BadNarrowOperatorError("unknown 'has' operand " + operand)

        if operand == "reaction":
            if self.msg_id_column.name == "message_id":
                # If the initial query uses `zerver_usermessage`
                check_col = literal_column("zerver_usermessage.message_id", Integer)
            else:
                # If the initial query doesn't use `zerver_usermessage`
                check_col = literal_column("zerver_message.id", Integer)
            exists_cond = (
                select(1)
                .select_from(table("zerver_reaction"))
                .where(check_col == literal_column("zerver_reaction.message_id", Integer))
                .exists()
            )
            return query.where(maybe_negate(exists_cond))

        col_name = "has_" + operand
        cond = column(col_name, Boolean)
        return query.where(maybe_negate(cond))

    def by_in(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        # This operator does not support is_web_public_query.
        assert not self.is_web_public_query
        assert self.user_profile is not None

        if operand == "home":
            conditions = exclude_muting_conditions(
                self.user_profile, [NarrowParameter(operator="in", operand="home")]
            )
            if conditions:
                return query.where(maybe_negate(and_(*conditions)))
            return query.where(maybe_negate(true()))
        elif operand == "all":
            return query

        raise BadNarrowOperatorError("unknown 'in' operand " + operand)

    def by_is(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        # Only `resolved` operand of this class supports is_web_public_query.
        if operand == "resolved":
            cond = get_resolved_topic_condition_sa()
            return query.where(maybe_negate(cond))

        assert not self.is_web_public_query
        assert self.user_profile is not None

        if operand in ["dm", "private"]:
            # "is:private" is a legacy alias for "is:dm"
            if maybe_negate is not_:
                self.check_not_both_channel_and_dm_narrow(
                    maybe_negate=lambda cond: cond, is_channel_narrow=True
                )
            else:
                self.check_not_both_channel_and_dm_narrow(maybe_negate, is_dm_narrow=True)
            cond = column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0
            return query.where(maybe_negate(cond))
        elif operand == "starred":
            cond = column("flags", Integer).op("&")(UserMessage.flags.starred.mask) != 0
            return query.where(maybe_negate(cond))
        elif operand == "unread":
            cond = column("flags", Integer).op("&")(UserMessage.flags.read.mask) == 0
            return query.where(maybe_negate(cond))
        elif operand == "mentioned":
            mention_flags_mask = (
                UserMessage.flags.mentioned.mask
                | UserMessage.flags.stream_wildcard_mentioned.mask
                | UserMessage.flags.topic_wildcard_mentioned.mask
                | UserMessage.flags.group_mentioned.mask
            )
            cond = column("flags", Integer).op("&")(mention_flags_mask) != 0
            return query.where(maybe_negate(cond))
        elif operand == "alerted":
            cond = column("flags", Integer).op("&")(UserMessage.flags.has_alert_word.mask) != 0
            return query.where(maybe_negate(cond))
        elif operand == "followed":
            cond = get_followed_topic_condition_sa(self.user_profile.id)
            return query.where(maybe_negate(cond))
        elif operand == "muted":
            # TODO: If we also have a channel operator, this could be
            # a lot more efficient if limited to only those muting
            # rules that appear in such channels.
            conditions = exclude_muting_conditions(
                self.user_profile, [NarrowParameter(operator="is", operand="muted")]
            )
            if conditions:
                return query.where(maybe_negate(not_(and_(*conditions))))

            # This is the case where no channels or topics were muted.
            return query.where(maybe_negate(false()))

        raise BadNarrowOperatorError("unknown 'is' operand " + operand)

    _alphanum = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")

    def by_channel(
        self, query: Select, operand: str | int, maybe_negate: ConditionTransform
    ) -> Select:
        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_channel_narrow=True)

        try:
            # Because you can see your own message history for
            # private channels you are no longer subscribed to, we
            # need get_stream_by_narrow_operand_access_unchecked here.
            channel = get_stream_by_narrow_operand_access_unchecked(operand, self.realm)

            if self.is_web_public_query and not channel.is_web_public:
                raise BadNarrowOperatorError("unknown web-public channel " + str(operand))
        except Stream.DoesNotExist:
            raise BadNarrowOperatorError("unknown channel " + str(operand))

        recipient_id = channel.recipient_id
        assert recipient_id is not None
        cond = column("recipient_id", Integer) == recipient_id
        return query.where(maybe_negate(cond))

    def by_channels(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_channel_narrow=True)

        if operand == "public":
            # Get all both subscribed and non-subscribed public channels
            # but exclude any private subscribed channels.
            recipient_queryset = get_public_streams_queryset(self.realm)
        elif operand == "web-public":
            recipient_queryset = get_web_public_streams_queryset(self.realm)
        else:
            raise BadNarrowOperatorError("unknown channels operand " + operand)

        recipient_ids = recipient_queryset.values_list("recipient_id", flat=True).order_by("id")
        cond = column("recipient_id", Integer).in_(recipient_ids)
        return query.where(maybe_negate(cond))

    def by_topic(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_channel_narrow=True)

        cond = topic_match_sa(operand)
        return query.where(maybe_negate(cond))

    def by_sender(
        self, query: Select, operand: str | int, maybe_negate: ConditionTransform
    ) -> Select:
        try:
            if isinstance(operand, str):
                sender = get_user_including_cross_realm(operand, self.realm)
            else:
                sender = get_user_by_id_in_realm_including_cross_realm(operand, self.realm)
        except UserProfile.DoesNotExist:
            raise BadNarrowOperatorError("unknown user " + str(operand))

        cond = column("sender_id", Integer) == literal(sender.id)
        return query.where(maybe_negate(cond))

    def by_near(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        return query

    def by_id(self, query: Select, operand: int | str, maybe_negate: ConditionTransform) -> Select:
        if not str(operand).isdigit() or int(operand) > Message.MAX_POSSIBLE_MESSAGE_ID:
            raise BadNarrowOperatorError("Invalid message ID")
        cond = self.msg_id_column == literal(operand)
        return query.where(maybe_negate(cond))

    def by_dm(
        self, query: Select, operand: str | Iterable[int], maybe_negate: ConditionTransform
    ) -> Select:
        # This operator does not support is_web_public_query.
        assert not self.is_web_public_query
        assert self.user_profile is not None

        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_dm_narrow=True)

        try:
            if isinstance(operand, str):
                email_list = operand.split(",")
                user_profiles = get_user_profiles(
                    emails=email_list,
                    realm=self.realm,
                )
            else:
                """
                This is where we handle passing a list of user IDs for the narrow, which is the
                preferred/cleaner API.
                """
                user_profiles = get_user_profiles_by_ids(
                    user_ids=operand,
                    realm=self.realm,
                )

            if user_profiles == []:
                return query.where(maybe_negate(false()))

            recipient = recipient_for_user_profiles(
                user_profiles=user_profiles,
                forwarded_mirror_message=False,
                forwarder_user_profile=None,
                sender=self.user_profile,
                allow_deactivated=True,
                create=False,
            )
        except (JsonableError, ValidationError):
            raise BadNarrowOperatorError("unknown user in " + str(operand))
        except DirectMessageGroup.DoesNotExist:
            # Group DM where direct message group doesn't exist
            return query.where(maybe_negate(false()))

        # Group direct message
        if recipient.type == Recipient.DIRECT_MESSAGE_GROUP:
            cond = column("recipient_id", Integer) == recipient.id
            return query.where(maybe_negate(cond))

        # 1:1 direct message
        other_participant = None

        # Find if another person is in direct message
        for user in user_profiles:
            if user.id != self.user_profile.id:
                other_participant = user

        # Direct message with another person
        if other_participant:
            # We need bidirectional direct messages with another person.
            # But Recipient.PERSONAL objects only encode the person who
            # received the message, and not the other participant in
            # the thread (the sender), we need to do a somewhat
            # complex query to get messages between these two users
            # with either of them as the sender.
            self_recipient_id = self.user_profile.recipient_id
            cond = and_(
                column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0,
                column("realm_id", Integer) == self.realm.id,
                or_(
                    and_(
                        column("sender_id", Integer) == other_participant.id,
                        column("recipient_id", Integer) == self_recipient_id,
                    ),
                    and_(
                        column("sender_id", Integer) == self.user_profile.id,
                        column("recipient_id", Integer) == recipient.id,
                    ),
                ),
            )
            return query.where(maybe_negate(cond))

        # Direct message with self
        cond = and_(
            column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0,
            column("realm_id", Integer) == self.realm.id,
            column("sender_id", Integer) == self.user_profile.id,
            column("recipient_id", Integer) == recipient.id,
        )
        return query.where(maybe_negate(cond))

    def _get_direct_message_group_recipients(self, other_user: UserProfile) -> set[int]:
        self_recipient_ids = [
            recipient_tuple["recipient_id"]
            for recipient_tuple in Subscription.objects.filter(
                user_profile=self.user_profile,
                recipient__type=Recipient.DIRECT_MESSAGE_GROUP,
            ).values("recipient_id")
        ]
        narrow_recipient_ids = [
            recipient_tuple["recipient_id"]
            for recipient_tuple in Subscription.objects.filter(
                user_profile=other_user,
                recipient__type=Recipient.DIRECT_MESSAGE_GROUP,
            ).values("recipient_id")
        ]

        return set(self_recipient_ids) & set(narrow_recipient_ids)

    def by_dm_including(
        self, query: Select, operand: str | int, maybe_negate: ConditionTransform
    ) -> Select:
        # This operator does not support is_web_public_query.
        assert not self.is_web_public_query
        assert self.user_profile is not None

        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_dm_narrow=True)

        try:
            if isinstance(operand, str):
                narrow_user_profile = get_user_including_cross_realm(operand, self.realm)
            else:
                narrow_user_profile = get_user_by_id_in_realm_including_cross_realm(
                    operand, self.realm
                )
        except UserProfile.DoesNotExist:
            raise BadNarrowOperatorError("unknown user " + str(operand))

        # "dm-including" when combined with the user's own ID/email as the operand
        # should return all group and 1:1 direct messages (including direct messages
        # with self), so the simplest query to get these messages is the same as "is:dm".
        if narrow_user_profile.id == self.user_profile.id:
            cond = column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0
            return query.where(maybe_negate(cond))

        # all direct messages including another person (group and 1:1)
        direct_message_group_recipient_ids = self._get_direct_message_group_recipients(
            narrow_user_profile
        )

        private_flag = column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0
        realm_match = column("realm_id", Integer) == self.realm.id
        direct_message_group_cond = column("recipient_id", Integer).in_(
            direct_message_group_recipient_ids
        )

        self_recipient_id = self.user_profile.recipient_id
        if self_recipient_id is not None and narrow_user_profile.recipient_id is not None:
            # See note above in `by_dm` about needing bidirectional messages
            # for direct messages with another person.
            dm_conditions = or_(
                and_(
                    column("sender_id", Integer) == narrow_user_profile.id,
                    column("recipient_id", Integer) == self_recipient_id,
                ),
                and_(
                    column("sender_id", Integer) == self.user_profile.id,
                    column("recipient_id", Integer) == narrow_user_profile.recipient_id,
                ),
                direct_message_group_cond,
            )
            cond = and_(private_flag, realm_match, dm_conditions)
        else:
            # If the user does not have a recipient_id, then they only rely on
            # direct group message for their conversations.
            cond = and_(private_flag, realm_match, direct_message_group_cond)
        return query.where(maybe_negate(cond))

    def by_group_pm_with(
        self, query: Select, operand: str | int, maybe_negate: ConditionTransform
    ) -> Select:
        # This operator does not support is_web_public_query.
        assert not self.is_web_public_query
        assert self.user_profile is not None

        self.check_not_both_channel_and_dm_narrow(maybe_negate, is_dm_narrow=True)

        try:
            if isinstance(operand, str):
                narrow_profile = get_user_including_cross_realm(operand, self.realm)
            else:
                narrow_profile = get_user_by_id_in_realm_including_cross_realm(operand, self.realm)
        except UserProfile.DoesNotExist:
            raise BadNarrowOperatorError("unknown user " + str(operand))

        recipient_ids = self._get_direct_message_group_recipients(narrow_profile)
        cond = and_(
            column("flags", Integer).op("&")(UserMessage.flags.is_private.mask) != 0,
            column("realm_id", Integer) == self.realm.id,
            column("recipient_id", Integer).in_(recipient_ids),
        )
        return query.where(maybe_negate(cond))

    def by_search(self, query: Select, operand: str, maybe_negate: ConditionTransform) -> Select:
        if settings.USING_PGROONGA:
            return self._by_search_pgroonga(query, operand, maybe_negate)
        else:
            return self._by_search_tsearch(query, operand, maybe_negate)

    def _by_search_pgroonga(
        self, query: Select, operand: str, maybe_negate: ConditionTransform
    ) -> Select:
        match_positions_character = func.pgroonga_match_positions_character
        query_extract_keywords = func.pgroonga_query_extract_keywords
        operand_escaped = func.escape_html(operand, type_=Text)
        keywords = query_extract_keywords(operand_escaped)
        query = query.add_columns(
            match_positions_character(column("rendered_content", Text), keywords).label(
                "content_matches"
            ),
            match_positions_character(
                func.escape_html(topic_column_sa(), type_=Text), keywords
            ).label("topic_matches"),
        )
        condition = column("search_pgroonga", Text).op("&@~")(operand_escaped)
        return query.where(maybe_negate(condition))

    def _by_search_tsearch(
        self, query: Select, operand: str, maybe_negate: ConditionTransform
    ) -> Select:
        tsquery = func.plainto_tsquery(literal("zulip.english_us_search"), literal(operand))
        query = query.add_columns(
            ts_locs_array(
                literal("zulip.english_us_search", Text), column("rendered_content", Text), tsquery
            ).label("content_matches"),
            # We HTML-escape the topic in PostgreSQL to avoid doing a server round-trip
            ts_locs_array(
                literal("zulip.english_us_search", Text),
                func.escape_html(topic_column_sa(), type_=Text),
                tsquery,
            ).label("topic_matches"),
        )

        # Do quoted string matching.  We really want phrase
        # search here so we can ignore punctuation and do
        # stemming, but there isn't a standard phrase search
        # mechanism in PostgreSQL
        for term in re.findall(r'"[^"]+"|\S+', operand):
            if term[0] == '"' and term[-1] == '"':
                term = term[1:-1]
                term = "%" + connection.ops.prep_for_like_query(term) + "%"
                cond: ClauseElement = or_(
                    column("content", Text).ilike(term),
                    and_(
                        topic_column_sa().ilike(term),
                        column("is_channel_message", Boolean),
                    ),
                )
                query = query.where(maybe_negate(cond))

        cond = column("search_tsvector", postgresql.TSVECTOR).op("@@")(tsquery)
        return query.where(maybe_negate(cond))


def ok_to_include_history(
    narrow: list[NarrowParameter] | None,
    user_profile: UserProfile | None,
    is_web_public_query: bool,
) -> bool:
    # There are occasions where we need to find Message rows that
    # have no corresponding UserMessage row, because the user is
    # reading a public channel that might include messages that
    # were sent while the user was not subscribed, but which they are
    # allowed to see.  We have to be very careful about constructing
    # queries in those situations, so this function should return True
    # only if we are 100% sure that we're gonna add a clause to the
    # query that narrows to a particular public channel on the user's realm.
    # If we screw this up, then we can get into a nasty situation of
    # polluting our narrow results with messages from other realms.

    # For web-public queries, we are always returning history.  The
    # analogues of the below channel access checks for whether channels
    # have is_web_public set and banning is operators in this code
    # path are done directly in NarrowBuilder.
    if is_web_public_query:
        assert user_profile is None
        return True

    assert user_profile is not None

    include_history = False
    if narrow is not None:
        for term in narrow:
