insights/insights/insightsql/property.py

import re
from collections.abc import Callable
from typing import Literal, Optional, TypeGuard, cast

from django.db import models
from django.db.models import Q
from django.db.models.functions.comparison import Coalesce

import re2
from pydantic import BaseModel
from rest_framework.exceptions import ValidationError

from insights.schema import (
    AccountCustomPropertyFilter,
    CohortPropertyFilter,
    DataWarehousePersonPropertyFilter,
    DataWarehousePropertyFilter,
    ElementPropertyFilter,
    EmptyPropertyFilter,
    ErrorTrackingIssueFilter,
    EventMetadataPropertyFilter,
    EventPropertyFilter,
    FeaturePropertyFilter,
    FilterLogicalOperator,
    FlagPropertyFilter,
    GroupPropertyFilter,
    InsightsQLPropertyFilter,
    LogEntryPropertyFilter,
    LogPropertyFilter,
    MetricPropertyFilter,
    PersonMetadataPropertyFilter,
    PersonPropertyFilter,
    PropertyGroupFilter,
    PropertyGroupFilterValue,
    PropertyOperator,
    RecordingPropertyFilter,
    RetentionEntity,
    RevenueAnalyticsPropertyFilter,
    SessionPropertyFilter,
    SpanPropertyFilter,
    WorkflowVariablePropertyFilter,
)

from insights.insightsql import ast
from insights.insightsql.base import AST
from insights.insightsql.constants import EXCEPTION_STRING_ARRAY_PROPERTIES
from insights.insightsql.database.models import BooleanDatabaseField
from insights.insightsql.database.schema.sessions import LAZY_SESSIONS_FIELDS
from insights.insightsql.errors import NotImplementedError, QueryError
from insights.insightsql.functions import find_insightsql_aggregation
from insights.insightsql.parser import CacheOrigin, parse_expr
from insights.insightsql.utils import map_virtual_properties
from insights.insightsql.visitor import CloningVisitor, TraversingVisitor, clone_expr

from insights.datastore.query_tagging import tag_contains_user_insightsql
from insights.constants import AUTOCAPTURE_EVENT, TREND_FILTER_TYPE_ACTIONS, PropertyOperatorType
from insights.models import Property, PropertyDefinition, Team
from insights.models.element import Element
from insights.models.event import Selector
from insights.models.property import PropertyGroup, ValueT
from insights.models.property.util import build_selector_regex
from insights.utils import get_from_dict_or_attr

from products.actions.backend.models.action import Action, ActionStepJSON
from products.cohorts.backend.models.cohort import Cohort
from products.data_tools.backend.models.join import DataWarehouseJoin
from products.event_definitions.backend.models.property_definition import PropertyType
from products.warehouse_sources.backend.facade.insightsql import get_view_or_table_by_name

# Top-level columns on the persons table that the `person_metadata` filter type can target.
# Keep in sync with the "person_metadata" group in
# frontend/src/taxonomy/core-filter-definitions-by-group.json,
# personMetadataPropertyDefinitions in frontend/src/models/propertyDefinitionsModel.ts,
# and the per-field injection in rust/feature-flags/src/flags/flag_matching_utils.rs.
# `test_person_metadata_fields_match_taxonomy` enforces the Python ↔ taxonomy half.
PERSON_METADATA_FIELDS = {"created_at"}


def parse_semver(value: str) -> tuple[str, str, str]:
    """
    Parse a semver string into (major, minor, patch) components.

    - Strips pre-release suffixes (e.g., -alpha.1)
    - Defaults missing components to "0" (e.g., 1.0 -> 1.0.0)

    Returns tuple of strings for direct use in version string construction.
    Raises ValueError if parsing fails.
    """
    # Strip pre-release suffix (everything after first hyphen)
    base_version = value.split("-")[0]

    parts = base_version.split(".")
    if len(parts) < 1 or not parts[0]:
        raise ValueError("Invalid semver format")

    major = parts[0]
    minor = parts[1] if len(parts) > 1 else "0"
    patch = parts[2] if len(parts) > 2 else "0"

    # Validate they're actually integers
    int(major), int(minor), int(patch)

    return (major, minor, patch)


# Anchored, strict-semver validator used to gate semver comparisons. We can't rely on
# `sortableSemver` returning NULL for invalid input because Datastore forbids
# `Nullable(Array(...))`, and an `Array(Nullable(Int64))` with a NULL element sorts
# as the *greatest* array — which would incorrectly include invalid versions in
# `>= filter` queries. Instead we gate every semver comparison on this regex
# matching the property side, so invalid values are dropped before the array
# comparison happens. Matches Rust `semver::Version::parse` (X.Y.Z, no leading
# zeros, optional 'v' prefix, optional pre-release / build suffix).
STRICT_SEMVER_REGEX = r"^\s*v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+][^\s]*)?\s*$"


def _gate_on_valid_semver(expr: ast.Expr, comparison: ast.Expr) -> ast.And:
    """Wrap a semver comparison so it only fires on rows where `expr` is strict semver."""
    return ast.And(
        exprs=[
            ast.Call(name="match", args=[expr, ast.Constant(value=STRICT_SEMVER_REGEX)]),
            comparison,
        ]
    )


def semver_range_compare(
    expr: ast.Expr,
    value: ast.Any,
    operator_name: str,
    bounds_calculator: Callable[[str], tuple[str, str]],
) -> ast.And:
    """
    Build a semver range comparison AST (lower_bound <= expr < upper_bound).

    Args:
        expr: The expression to compare (e.g., person.properties.app_version)
        value: The semver value from the filter
        operator_name: Name for error messages (e.g., "Tilde", "Caret", "Wildcard")
        bounds_calculator: Function that takes the value and returns (lower_bound, upper_bound)

    Returns:
        AST node representing: match(expr, valid_semver) AND sortableSemver(expr) >= sortableSemver(lower) AND sortableSemver(expr) < sortableSemver(upper)
    """
    if not isinstance(value, str):
        raise QueryError(f"{operator_name} operator requires a semver string value")

    try:
        lower_bound, upper_bound = bounds_calculator(value)
    except (ValueError, IndexError):
        raise QueryError(f"{operator_name} operator requires a valid semver string (e.g., '1.2.3')")

    return ast.And(
        exprs=[
            ast.Call(name="match", args=[expr, ast.Constant(value=STRICT_SEMVER_REGEX)]),
            ast.CompareOperation(
                op=ast.CompareOperationOp.GtEq,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=lower_bound)]),
            ),
            ast.CompareOperation(
                op=ast.CompareOperationOp.Lt,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=upper_bound)]),
            ),
        ]
    )


def _tilde_bounds(value: str) -> tuple[str, str]:
    """
    ~1.2.3 means >=1.2.3 <1.3.0 (allows patch-level changes)
    ~1 means >=1.0.0 <2.0.0 (bare major: allows minor+patch changes)
    """
    major, minor, patch = parse_semver(value)
    parts = value.split("-")[0].split(".")
    if len(parts) < 2:
        return f"{major}.0.0", f"{int(major) + 1}.0.0"
    next_minor = str(int(minor) + 1)
    return f"{major}.{minor}.{patch}", f"{major}.{next_minor}.0"


def _caret_bounds(value: str) -> tuple[str, str]:
    """
    Caret operator follows semver spec:
    ^1.2.3 means >=1.2.3 <2.0.0
    ^0.2.3 means >=0.2.3 <0.3.0
    ^0.0.3 means >=0.0.3 <0.0.4
    The leftmost non-zero component determines the upper bound.
    """
    major, minor, patch = parse_semver(value)
    lower_bound = f"{major}.{minor}.{patch}"

    if int(major) > 0:
        upper_bound = f"{int(major) + 1}.0.0"
    elif int(minor) > 0:
        upper_bound = f"0.{int(minor) + 1}.0"
    else:
        upper_bound = f"0.0.{int(patch) + 1}"

    return lower_bound, upper_bound


def _wildcard_bounds(value: str) -> tuple[str, str]:
    """
    Wildcard matching:
    1.* means >=1.0.0 <2.0.0
    1.2.* means >=1.2.0 <1.3.0
    1.2.3.* means >=1.2.3.0 <1.2.4.0
    """
    # Remove trailing .* if present
    value = value.rstrip(".*")
    if not value:
        raise ValueError("Invalid wildcard pattern")

    # Strip pre-release suffix before counting parts
    base_value = value.split("-")[0]
    parts = base_value.split(".")

    if len(parts) == 1:
        major = parts[0]
        int(major)  # Validate
        return f"{major}.0.0", f"{int(major) + 1}.0.0"
    elif len(parts) == 2:
        major, minor = parts[0], parts[1]
        int(major), int(minor)  # Validate
        return f"{major}.{minor}.0", f"{major}.{int(minor) + 1}.0"
    else:
        major, minor, patch = parts[0], parts[1], parts[2]
        int(major), int(minor), int(patch)  # Validate
        return f"{major}.{minor}.{patch}.0", f"{major}.{minor}.{int(patch) + 1}.0"


GROUP_KEY_PATTERN = re.compile(r"^\$group_[0-4]$")


def _stringify_group_key_value(value: object) -> str | list[str]:
    """Group keys ($group_0–$group_4) are always stored as strings. A numeric filter
    value would produce equals(<string column>, <number>), which Datastore rejects
    with NO_COMMON_TYPE — so coerce a group-key filter value to its string form."""

    def _scalar(v: object) -> str:
        # An integer-valued float ('13.0') stringifies to the plain integer ('13')
        if isinstance(v, float) and v.is_integer():
            return str(int(v))
        return str(v)

    if isinstance(value, list):
        return [_scalar(v) for v in value]
    return _scalar(value)


def has_aggregation(expr: AST) -> bool:
    finder = AggregationFinder()
    finder.visit(expr)
    return finder.has_aggregation


class AggregationFinder(TraversingVisitor):
    def __init__(self):
        super().__init__()
        self.has_aggregation = False

    def visit(self, node):
        if self.has_aggregation:
            return
        else:
            super().visit(node)

    def visit_select_query(self, node: ast.SelectQuery):
        # don't care about aggregations in subqueries
        pass

    def visit_call(self, node: ast.Call):
        if find_insightsql_aggregation(node.name):
            self.has_aggregation = True
        else:
            for arg in node.args:
                self.visit(arg)


def _handle_bool_values(value: ValueT, expr: ast.Expr, property: Property, team: Team) -> ValueT | bool:
    if value is True:
        value = "true"
    elif value is False:
        value = "false"

    if value != "true" and value != "false":
        return value

    # Virtual event properties (e.g. $virt_is_bot) don't have PropertyDefinition
    # records, so we check the taxonomy directly for boolean type
    if property.key and property.key.startswith("$virt_"):
        from insights.taxonomy.taxonomy import CORE_FILTER_DEFINITIONS_BY_GROUP

        for group_defs in CORE_FILTER_DEFINITIONS_BY_GROUP.values():
            prop_def = group_defs.get(property.key)
            if prop_def and prop_def.get("type") == "Boolean":
                return value == "true"
        return value

    if property.type == "person":
        property_types = PropertyDefinition.objects.alias(
            effective_project_id=Coalesce("project_id", "team_id", output_field=models.BigIntegerField())
        ).filter(
            effective_project_id=team.project_id,
            name=property.key,
            type=PropertyDefinition.Type.PERSON,
        )
    elif property.type == "group":
        property_types = PropertyDefinition.objects.alias(
            effective_project_id=Coalesce("project_id", "team_id", output_field=models.BigIntegerField())
        ).filter(
            effective_project_id=team.project_id,
            name=property.key,
            type=PropertyDefinition.Type.GROUP,
            group_type_index=property.group_type_index,
        )
    elif property.type == "data_warehouse_person_property":
        if not isinstance(expr, ast.Field):
            raise Exception(f"Requires a Field expression")

        key = expr.chain[-2]

        # TODO: pass id of table item being filtered on instead of searching through joins
        current_join: DataWarehouseJoin | None = (
            DataWarehouseJoin.objects.filter(Q(deleted__isnull=True) | Q(deleted=False))
            .filter(team=team, source_table_name="persons", field_name=key)
            .first()
        )

        if not current_join:
            raise Exception(f"Could not find join for key {key}")

        prop_type = None

        table_or_view = get_view_or_table_by_name(team, current_join.joining_table_name)
        if table_or_view and table_or_view.columns is not None:
            prop_type_dict = table_or_view.columns.get(property.key, None)
            if prop_type_dict is not None:
                prop_type = prop_type_dict.get("insightsql")

        if not table_or_view:
            raise Exception(f"Could not find table or view for key {key}")

        if prop_type == "BooleanDatabaseField":
            if value == "true":
                value = True
            if value == "false":
                value = False

        return value

    elif property.type == "session":
        field_definition = LAZY_SESSIONS_FIELDS.get(property.key)
        if isinstance(field_definition, BooleanDatabaseField):
            if value == "true":
                return True
            if value == "false":
                return False
        return value

    else:
        property_types = PropertyDefinition.objects.alias(
            effective_project_id=Coalesce("project_id", "team_id", output_field=models.BigIntegerField())
        ).filter(
            effective_project_id=team.project_id,
            name=property.key,
            type=PropertyDefinition.Type.EVENT,
        )
    property_type = property_types[0].property_type if len(property_types) > 0 else None

    if property_type == PropertyType.Boolean:
        if value == "true":
            return True
        if value == "false":
            return False
    return value


def _resolve_date_value(value: ValueT, team: Team) -> ValueT:
    """Resolve a date value for IS_DATE_* operators.

    Relative dates (e.g. ``-7d``, ``-10m`` for months, ``-10M`` for minutes) are
    resolved against the team timezone and rendered as a naive ``YYYY-MM-DD HH:MM:SS``
    wall-clock string — the printer's constant-folding fast-path then lowers that
    to ``toDateTime(str, <team_tz>)``, which interprets it in the team timezone.
    Because ``relative_date_parse`` already computed the wall clock in the team
    timezone, the round-trip is correct.

    Absolute values are returned untouched. In particular, ISO 8601 strings with
    an explicit ``T``/``Z`` (e.g. ``2026-03-19T14:00:00Z``) are preserved so that
    Datastore's ``parseDateTime64BestEffort`` — which honors the embedded offset
    — can parse them to the correct UTC moment regardless of the team's timezone.
    Stripping the ``Z`` and fast-pathing to ``toDateTime`` would silently reinterpret
    the value in the team timezone and shift it by the offset.
    """
    if not isinstance(value, str):
        return value

    relative_regex = r"^-?[0-9]+[hdwmqysHDWMQY]"
    if re.match(relative_regex, value):
        from insights.utils import relative_date_parse

        resolved = relative_date_parse(value, team.timezone_info)
        return resolved.strftime("%Y-%m-%d %H:%M:%S")

    return value


def _force_datetime(expr: ast.Expr) -> ast.Expr:
    """Coerce ``expr`` to DateTime for chronological IS_DATE_* comparison.

    PropertySwapper only wraps the LHS of a comparison in ``toDateTime()``
    when a DateTime PropertyDefinition exists for the property. Without one,
    the LHS stays as a raw JSON-extracted String, and a String-vs-String
    comparison becomes lexicographic — which silently diverges from
    chronological order when one side uses the ISO 8601 ``T``/``Z`` form and
    the other the normalized MySQL form (``'T'`` = 0x54 sorts after
    ``' '`` = 0x20).

    Wrapping both sides in InsightsQL's ``toDateTime`` forces Datastore to parse
    and compare as DateTime regardless of the underlying column type.

    The ``toString`` hop on non-constant expressions is load-bearing: when
    PropertySwapper *does* wrap the LHS (DateTime PropertyDefinition exists),
    InsightsQL's overload resolution for the outer ``toDateTime`` cannot see
    through PropertySwapper's freshly created Call node and falls back to
    the default ``parseDateTime64BestEffortOrNull`` mapping. That function
    rejects DateTime64 input (``ILLEGAL_TYPE_OF_ARGUMENT``), so we must
    stringify first. String constants already skip the hop because
    ``toString('x')`` is pointless and omitting it lets the printer apply
    its ``parseDateTime64BestEffortOrNull`` → ``toDateTime`` constant
    folding optimization.
    """
    if isinstance(expr, ast.Constant) and isinstance(expr.value, str):
        return ast.Call(name="toDateTime", args=[expr])
    return ast.Call(
        name="toDateTime",
        args=[ast.Call(name="toString", args=[expr])],
    )


def _validate_between_values(value: ValueT, operator: PropertyOperator) -> TypeGuard[list[str]]:
    if not isinstance(value, list) or len(value) != 2:
        raise QueryError(f"{operator} operator requires a two-element array [min, max]")
    try:
        if float(value[0]) > float(value[1]):
            raise QueryError(f"{operator} operator requires min value to be less than or equal to max value")
    except (ValueError, TypeError):
        raise QueryError(f"{operator} operator requires numeric values")
    return True


def _multi_search_found(search_call: ast.Call) -> ast.CompareOperation:
    """Create comparison operation to check if multiSearchAnyCaseInsensitive found a match."""
    return ast.CompareOperation(op=ast.CompareOperationOp.Gt, left=search_call, right=ast.Constant(value=0))


def _multi_search_not_found(search_call: ast.Call) -> ast.CompareOperation:
    """Create comparison operation to check if multiSearchAnyCaseInsensitive did not find a match."""
    return ast.CompareOperation(op=ast.CompareOperationOp.Eq, left=search_call, right=ast.Constant(value=0))


def _create_multi_search_call(expr: ast.Expr, value: list) -> ast.Call:
    """Create a multiSearchAnyCaseInsensitive call for the given expression and values."""
    return ast.Call(
        name="multiSearchAnyCaseInsensitive",
        args=[
            ast.Call(name="toString", args=[expr]),
            ast.Array(exprs=[ast.Constant(value=str(v)) for v in value]),
        ],
    )


def _validate_regex(value: ValueT) -> None:
    """Reject an invalid regular expression with a clear user-facing error rather
    than letting Datastore fail the whole query with CANNOT_COMPILE_REGEXP. The
    same RE2 engine Datastore uses validates the pattern here."""
    if not isinstance(value, str):
        return
    try:
        re2.compile(value)
    except re2.error as err:
        raise QueryError(f"Invalid regular expression: '{value}'") from err


def _expr_to_compare_op(
    expr: ast.Expr, value: ValueT, operator: PropertyOperator, property: Property, is_json_field: bool, team: Team
) -> ast.Expr:
    if operator == PropertyOperator.IS_SET:
        return ast.CompareOperation(
            op=ast.CompareOperationOp.NotEq,
            left=expr,
            right=ast.Constant(value=None),
        )
    elif operator == PropertyOperator.IS_NOT_SET:
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Eq,
            left=expr,
            right=ast.Constant(value=None),
        )
    elif operator == PropertyOperator.ICONTAINS:
        if isinstance(value, list) and len(value) > 1:
            # Multiple values: use Datastore's multiSearchAnyCaseInsensitive for efficient searching
            return _multi_search_found(_create_multi_search_call(expr, value))
        else:
            # Single value (or single-element array): keep existing ILIKE logic for backward compatibility
            single_value = value[0] if isinstance(value, list) and len(value) == 1 else value
            return ast.CompareOperation(
                op=ast.CompareOperationOp.ILike,
                left=ast.Call(name="toString", args=[expr]),
                right=ast.Constant(value=f"%{single_value}%"),
            )
    elif operator == PropertyOperator.NOT_ICONTAINS:
        if isinstance(value, list) and len(value) > 1:
            # Multiple values: use Datastore's multiSearchAnyCaseInsensitive with negation
            return _multi_search_not_found(_create_multi_search_call(expr, value))
        else:
            # Single value (or single-element array): keep existing NOT ILIKE logic for backward compatibility
            single_value = value[0] if isinstance(value, list) and len(value) == 1 else value
            return ast.CompareOperation(
                op=ast.CompareOperationOp.NotILike,
                left=ast.Call(name="toString", args=[expr]),
                right=ast.Constant(value=f"%{single_value}%"),
            )
    elif operator in (PropertyOperator.STARTS_WITH, PropertyOperator.NOT_STARTS_WITH):
        single_value = value[0] if isinstance(value, list) and len(value) == 1 else value
        return ast.CompareOperation(
            op=ast.CompareOperationOp.ILike
            if operator == PropertyOperator.STARTS_WITH
            else ast.CompareOperationOp.NotILike,
            left=ast.Call(name="toString", args=[expr]),
            right=ast.Constant(value=f"{single_value}%"),
        )
    elif operator in (PropertyOperator.ENDS_WITH, PropertyOperator.NOT_ENDS_WITH):
        single_value = value[0] if isinstance(value, list) and len(value) == 1 else value
        return ast.CompareOperation(
            op=ast.CompareOperationOp.ILike
            if operator == PropertyOperator.ENDS_WITH
            else ast.CompareOperationOp.NotILike,
            left=ast.Call(name="toString", args=[expr]),
            right=ast.Constant(value=f"%{single_value}"),
        )
    elif operator == PropertyOperator.ICONTAINS_MULTI:
        # Always expect multiple values for multi-contains operator
        if isinstance(value, list):
            values_list = value
        else:
            values_list = cast(list, [value])
        return _multi_search_found(_create_multi_search_call(expr, values_list))
    elif operator == PropertyOperator.NOT_ICONTAINS_MULTI:
        # Always expect multiple values for multi-not-contains operator
        if isinstance(value, list):
            values_list = value
        else:
            values_list = cast(list, [value])
        return _multi_search_not_found(_create_multi_search_call(expr, values_list))
    elif operator == PropertyOperator.REGEX:
        _validate_regex(value)
        return ast.Call(
            name="ifNull",
            args=[
                ast.Call(name="match", args=[ast.Call(name="toString", args=[expr]), ast.Constant(value=value)]),
                ast.Constant(value=0),
            ],
        )
    elif operator == PropertyOperator.NOT_REGEX:
        _validate_regex(value)
        return ast.Call(
            name="ifNull",
            args=[
                ast.Call(
                    name="not",
                    args=[
                        ast.Call(name="match", args=[ast.Call(name="toString", args=[expr]), ast.Constant(value=value)])
                    ],
                ),
                ast.Constant(value=1),
            ],
        )
    elif operator == PropertyOperator.EXACT:
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Eq,
            left=expr,
            right=ast.Constant(value=_handle_bool_values(value, expr, property, team)),
        )
    elif operator == PropertyOperator.IS_DATE_EXACT:
        assert isinstance(value, str)
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Eq,
            left=_force_datetime(expr),
            right=_force_datetime(ast.Constant(value=_resolve_date_value(value, team))),
        )
    elif operator == PropertyOperator.IS_NOT:
        return ast.CompareOperation(
            op=ast.CompareOperationOp.NotEq,
            left=expr,
            right=ast.Constant(value=_handle_bool_values(value, expr, property, team)),
        )
    elif operator == PropertyOperator.LT:
        return ast.CompareOperation(op=ast.CompareOperationOp.Lt, left=expr, right=ast.Constant(value=value))
    elif operator == PropertyOperator.IS_DATE_BEFORE:
        assert isinstance(value, str)
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Lt,
            left=_force_datetime(expr),
            right=_force_datetime(ast.Constant(value=_resolve_date_value(value, team))),
        )
    elif operator == PropertyOperator.GT:
        return ast.CompareOperation(op=ast.CompareOperationOp.Gt, left=expr, right=ast.Constant(value=value))
    elif operator == PropertyOperator.IS_DATE_AFTER:
        assert isinstance(value, str)
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Gt,
            left=_force_datetime(expr),
            right=_force_datetime(ast.Constant(value=_resolve_date_value(value, team))),
        )
    elif operator == PropertyOperator.LTE or operator == PropertyOperator.MAX:
        return ast.CompareOperation(op=ast.CompareOperationOp.LtEq, left=expr, right=ast.Constant(value=value))
    elif operator == PropertyOperator.GTE or operator == PropertyOperator.MIN:
        return ast.CompareOperation(op=ast.CompareOperationOp.GtEq, left=expr, right=ast.Constant(value=value))
    elif operator == PropertyOperator.BETWEEN:
        _validate_between_values(value, operator)
        assert isinstance(value, list)
        return ast.And(
            exprs=[
                ast.CompareOperation(op=ast.CompareOperationOp.GtEq, left=expr, right=ast.Constant(value=value[0])),
                ast.CompareOperation(op=ast.CompareOperationOp.LtEq, left=expr, right=ast.Constant(value=value[1])),
            ]
        )
    elif operator == PropertyOperator.NOT_BETWEEN:
        _validate_between_values(value, operator)
        assert isinstance(value, list)
        return ast.Or(
            exprs=[
                ast.CompareOperation(op=ast.CompareOperationOp.Lt, left=expr, right=ast.Constant(value=value[0])),
                ast.CompareOperation(op=ast.CompareOperationOp.Gt, left=expr, right=ast.Constant(value=value[1])),
            ]
        )
    elif operator == PropertyOperator.IS_CLEANED_PATH_EXACT:
        return ast.CompareOperation(
            op=ast.CompareOperationOp.Eq,
            left=apply_path_cleaning(expr, team),
            right=apply_path_cleaning(ast.Constant(value=value), team),
        )
    elif operator == PropertyOperator.IN_ or operator == PropertyOperator.NOT_IN:
        if not isinstance(value, list):
            raise Exception("IN and NOT IN operators require a list of values")
        op = ast.CompareOperationOp.NotIn if operator == PropertyOperator.NOT_IN else ast.CompareOperationOp.In
        return ast.CompareOperation(op=op, left=expr, right=ast.Array(exprs=[ast.Constant(value=v) for v in value]))
    elif operator == PropertyOperator.SEMVER_EQ:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.Eq,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_NEQ:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.NotEq,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_GT:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.Gt,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_GTE:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.GtEq,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_LT:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.Lt,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_LTE:
        return _gate_on_valid_semver(
            expr,
            ast.CompareOperation(
                op=ast.CompareOperationOp.LtEq,
                left=ast.Call(name="sortableSemver", args=[expr]),
                right=ast.Call(name="sortableSemver", args=[ast.Constant(value=value)]),
            ),
        )
    elif operator == PropertyOperator.SEMVER_TILDE:
        return semver_range_compare(expr, value, "Tilde", _tilde_bounds)
    elif operator == PropertyOperator.SEMVER_CARET:
        return semver_range_compare(expr, value, "Caret", _caret_bounds)
    elif operator == PropertyOperator.SEMVER_WILDCARD:
        return semver_range_compare(expr, value, "Wildcard", _wildcard_bounds)
    else:
        raise NotImplementedError(f"PropertyOperator {operator} not implemented")


def apply_path_cleaning(path_expr: ast.Expr, team: Team) -> ast.Expr:
    if not team.path_cleaning_filters:
        return path_expr

    for replacement in team.path_cleaning_filter_models():
        path_expr = ast.Call(
            name="replaceRegexpAll",
            args=[
                path_expr,
                ast.Constant(value=replacement.regex),
                ast.Constant(value=replacement.alias),
            ],
        )

    return path_expr


def property_to_expr(
    property: (
        list
        | dict
        | PropertyGroup
        | PropertyGroupFilter
        | PropertyGroupFilterValue
        | Property
        | ast.Expr
        | EventPropertyFilter
        | PersonPropertyFilter
        | ElementPropertyFilter
        | SessionPropertyFilter
        | EventMetadataPropertyFilter
        | PersonMetadataPropertyFilter
        | RevenueAnalyticsPropertyFilter
        | AccountCustomPropertyFilter
        | CohortPropertyFilter
        | RecordingPropertyFilter
        | LogEntryPropertyFilter
        | GroupPropertyFilter
        | FeaturePropertyFilter
        | FlagPropertyFilter
        | InsightsQLPropertyFilter
        | EmptyPropertyFilter
        | DataWarehousePropertyFilter
        | DataWarehousePersonPropertyFilter
        | ErrorTrackingIssueFilter
        | LogPropertyFilter
        | MetricPropertyFilter
        | SpanPropertyFilter
        | WorkflowVariablePropertyFilter
    ),
    team: Team,
    scope: Literal[
        "event", "person", "group", "session", "replay", "replay_entity", "revenue_analytics", "log_resource"
    ] = "event",
    strict: bool = False,
) -> ast.Expr:
    if isinstance(property, dict):
        try:
            property = Property(**property)
        # The property was saved as an incomplete object. Instead of crashing the entire query, pretend it's not there.
        # TODO: revert this when removing legacy insights?
        except ValueError:
            if strict:
                raise
            return ast.Constant(value=1)
        except TypeError:
            if strict:
                raise
            return ast.Constant(value=1)
    elif isinstance(property, list):
        properties = [property_to_expr(p, team, scope, strict=strict) for p in property]
        if len(properties) == 0:
            return ast.Constant(value=1)
        if len(properties) == 1:
            return properties[0]
        return ast.And(exprs=properties)
    elif isinstance(property, Property):
        pass
    elif isinstance(property, ast.Expr):
        return clone_expr(property)
    elif (
        isinstance(property, PropertyGroup)
        or isinstance(property, PropertyGroupFilter)
        or isinstance(property, PropertyGroupFilterValue)
    ):
        if (
            isinstance(property, PropertyGroup)
            and property.type != PropertyOperatorType.AND
            and property.type != PropertyOperatorType.OR
        ):
            raise QueryError(f'PropertyGroup of unknown type "{property.type}"')
        if (
            (isinstance(property, PropertyGroupFilter) or isinstance(property, PropertyGroupFilterValue))
            and property.type != FilterLogicalOperator.AND_
            and property.type != FilterLogicalOperator.OR_
        ):
            raise QueryError(f'PropertyGroupFilter of unknown type "{property.type}"')

        if len(property.values) == 0:
            return ast.Constant(value=1)
        if len(property.values) == 1:
            return property_to_expr(property.values[0], team, scope, strict=strict)

        if property.type == PropertyOperatorType.AND or property.type == FilterLogicalOperator.AND_:
            return ast.And(exprs=[property_to_expr(p, team, scope, strict=strict) for p in property.values])
        else:
            return ast.Or(exprs=[property_to_expr(p, team, scope, strict=strict) for p in property.values])
    elif isinstance(property, EmptyPropertyFilter):
        return ast.Constant(value=1)
    elif isinstance(property, FlagPropertyFilter):
        # Flag dependencies are evaluated at the API layer, not in InsightsQL.
        # They should never reach this point, but we handle them gracefully
        # to satisfy type checking since FlagPropertyFilter is part
        # of the AnyPropertyFilter union used throughout the codebase.
        # Return a neutral filter that doesn't affect the query.
        return ast.Constant(value=1)
    elif isinstance(property, BaseModel):
        try:
            property = Property(**property.dict())
        except ValueError:
            if strict:
                raise
            # The property was saved as an incomplete object. Instead of crashing the entire query, pretend it's not there.
            return ast.Constant(value=1)
    else:
        raise QueryError(f"property_to_expr with property of type {type(property).__name__} not implemented")

    if property.type == "flag":
        # Flag dependencies are evaluated at flag-matching time and can't be expressed
        # in InsightsQL — return a neutral filter, mirroring the FlagPropertyFilter handling above.
        return ast.Constant(value=1)
    elif property.type == "insightsql":
        tag_contains_user_insightsql()
        return parse_expr(property.key, cache_origin=CacheOrigin.USER)
    elif property.type == "event_metadata" and scope == "group" and GROUP_KEY_PATTERN.match(property.key) is not None:
        group_type_index = property.key.split("_")[1]
        operator = cast(Optional[PropertyOperator], property.operator) or PropertyOperator.EXACT
        value = property.value
        if isinstance(property.value, list):
            if len(property.value) > 1:
                raise QueryError(f"The '{property.key}' property filter only supports one value in 'group' scope")
            value = property.value[0]

        value = _stringify_group_key_value(value)

        # For groups table, $group_N filters should match both index and key
        # index should equal N, and key should match the value
        index_condition = ast.CompareOperation(
            op=ast.CompareOperationOp.Eq,
            left=ast.Field(chain=["index"]),
            right=ast.Constant(value=int(group_type_index)),
        )

        key_condition = _expr_to_compare_op(
            expr=ast.Field(chain=["key"]),
            value=value,
            operator=operator,
            property=property,
            is_json_field=False,
            team=team,
        )

        return ast.And(exprs=[key_condition, index_condition])
    elif (
        property.type == "event"
        or property.type == "event_metadata"
        or property.type == "feature"
        or property.type == "person"
        or property.type == "person_metadata"
        or property.type == "group"
        or property.type == "behavioral"
        or property.type == "data_warehouse"
        or property.type == "data_warehouse_person_property"
        or property.type == "session"
        or property.type == "recording"
        or property.type == "log_entry"
        or property.type == "error_tracking_issue"
        or property.type == "log"
        or property.type == "log_attribute"
        or property.type == "log_resource_attribute"
        or property.type == "span"
        or property.type == "span_attribute"
        or property.type == "span_resource_attribute"
        or property.type == "revenue_analytics"
        or property.type == "account_custom_property"
        or property.type == "workflow_variable"
    ):
        if (
            (scope == "person" and property.type != "person" and property.type != "person_metadata")
            or (scope == "session" and property.type != "session")
            or (scope != "event" and property.type == "event_metadata")
            or (scope == "revenue_analytics" and property.type != "revenue_analytics")
            or (property.type == "revenue_analytics" and scope != "revenue_analytics")
            # Account custom properties resolve inside customer analytics queries only;
            # no generic InsightsQL scope can join them, so fail loudly instead of no-oping.
            or property.type == "account_custom_property"
        ):
            raise QueryError(f"The '{property.type}' property filter does not work in '{scope}' scope")

        if property.type == "person_metadata" and property.key not in PERSON_METADATA_FIELDS:
            raise QueryError(f"Unsupported person_metadata field: '{property.key}'")
        operator = cast(Optional[PropertyOperator], property.operator) or PropertyOperator.EXACT
        value = property.value

        if property.key and GROUP_KEY_PATTERN.match(str(property.key)):
            value = _stringify_group_key_value(value)

        if property.type == "person" and property.key == "distinct_id":
            # distinct_id is not stored in person.properties.
            # - In event scope, events.distinct_id is a real column — no join needed.
            # - In person scope, persons.pdi.distinct_id resolves via the lazy join on persons.
            # Routing through `events.person.pdi` would break under person-on-events mode,
            # where `events.person` is rebound to the `poe` virtual table which has no `pdi` field.
            if scope == "event":
                chain = []
            elif scope == "person":
                chain = ["pdi"]
            else:
                chain = ["person", "pdi"]
        elif property.type == "person" and scope != "person":
            chain = ["person", "properties"]
        elif property.type == "person_metadata":
            chain = ["person"] if scope != "person" else []
        elif property.type == "event" and scope == "replay_entity":
            chain = ["events", "properties"]
        elif property.type == "session" and scope == "replay_entity":
            chain = ["events", "session"]
        elif property.type == "data_warehouse":
            if not isinstance(property.key, str):
                raise QueryError("Data warehouse property filter value must be a string")
            else:
                split = property.key.split(".")
                chain = split[:-1]
                property.key = split[-1]

            if isinstance(value, list) and len(value) > 1:
                field = ast.Field(chain=[*chain, property.key])
                exprs = [
                    _expr_to_compare_op(
                        expr=field,
                        value=v,
                        operator=operator,
                        team=team,
                        property=property,
                        is_json_field=False,
                    )
                    for v in value
                ]
                if (
                    operator == PropertyOperator.NOT_ICONTAINS
                    or operator == PropertyOperator.NOT_REGEX
                    or operator == PropertyOperator.IS_NOT
                    or operator == PropertyOperator.NOT_STARTS_WITH
                    or operator == PropertyOperator.NOT_ENDS_WITH
                ):
                    return ast.And(exprs=exprs)
                return ast.Or(exprs=exprs)
        elif property.type == "data_warehouse_person_property":
            if isinstance(property.key, str):
                table, key = property.key.split(".")
                chain = ["person", table]
                property.key = key
            else:
                raise QueryError("Data warehouse person property filter value must be a string")
        elif property.type == "group" and scope != "group":
            chain = [f"group_{property.group_type_index}", "properties"]
        elif property.type == "session" and scope in ["event", "replay"]:
            chain = ["session"]
        elif property.type == "session" and scope == "session":
            chain = ["sessions"]
        elif property.type in ["recording", "data_warehouse", "log_entry", "event_metadata"]:
            chain = []
        elif property.type == "log":
            chain = [property.key]
            property.key = ""
        elif property.type == "span":
            chain = [property.key]
            property.key = ""
        elif scope == "log_resource":
            # log resource attributes are stored in a separate table as `attribute_key` and `attribute_value`
            # columns. The `attribute_key` filter needs to be added separately outside of property_to_expr
            chain = ["attribute_value"]
            property.key = ""
        elif property.type == "log_attribute":
            chain = ["attributes"]
        elif property.type == "log_resource_attribute":
            chain = ["resource_attributes"]
        elif property.type == "span_attribute":
            chain = ["attributes"]
        elif property.type == "span_resource_attribute":
            chain = ["resource_attributes"]
        elif property.type == "revenue_analytics":
            *chain, property.key = property.key.split(".")
        elif property.type == "workflow_variable":
            chain = ["variables"]
        else:
            chain = ["properties"]

        # We pretend elements chain is a property, but it is actually a column on the events table
        if chain == ["properties"] and property.key == "$elements_chain":
            field = ast.Field(chain=["elements_chain"])
        elif property.key == "":
            field = ast.Field(chain=[*chain])
        else:
            field = ast.Field(chain=[*chain, property.key])

        expr: ast.Expr = map_virtual_properties(field)

        if property.type == "recording" and property.key in ("snapshot_source", "snapshot_library"):
            expr = ast.Call(name="argMinMerge", args=[field])

        is_visited_page_property = property.type == "recording" and property.key == "visited_page"
        if is_visited_page_property:
            # Use the all_urls array field to filter for pages visited during recording.
            all_urls_field = ast.Call(name="groupUniqArrayArray", args=[ast.Field(chain=["all_urls"])])

        is_exception_string_array_property = (
            property.type == "event" and property.key in EXCEPTION_STRING_ARRAY_PROPERTIES
        )

        if is_exception_string_array_property:
            # if materialized these columns will be strings so we need to extract them
            extracted_field = ast.Call(
                name="JSONExtract",
                args=[
                    ast.Call(name="ifNull", args=[field, ast.Constant(value="")]),
                    ast.Constant(value="Array(String)"),
                ],
            )

        if isinstance(value, list) and operator not in (
            PropertyOperator.BETWEEN,
            PropertyOperator.NOT_BETWEEN,
            PropertyOperator.ICONTAINS,
            PropertyOperator.NOT_ICONTAINS,
            # starts_with/ends_with intentionally excluded: no Datastore anchored multi-search
            # primitive exists, so multi-value use falls through to per-value ILIKE scans below.
        ):
            if len(value) == 0:
                return ast.Constant(value=1)
            elif len(value) == 1:
                value = value[0]
            else:
                if operator in (
                    PropertyOperator.EXACT,
                    PropertyOperator.IS_NOT,
                    PropertyOperator.IN_,
                    PropertyOperator.NOT_IN,
                ):
                    op = (
                        ast.CompareOperationOp.In
                        if operator in (PropertyOperator.EXACT, PropertyOperator.IN_)
                        else ast.CompareOperationOp.NotIn
                    )

                    left = (
                        ast.Field(chain=["v"])
                        if (is_exception_string_array_property or is_visited_page_property)
                        else expr
                    )
                    compare_op = ast.CompareOperation(
                        op=op, left=left, right=ast.Tuple(exprs=[ast.Constant(value=v) for v in value])
                    )

                    if is_exception_string_array_property:
                        return parse_expr(
                            "arrayExists(v -> {compare_op}, {field})",
                            {
                                "compare_op": compare_op,
                                "field": extracted_field,
                            },
                        )
                    elif is_visited_page_property:
                        return parse_expr(
                            "arrayExists(v -> {compare_op}, {field})",
                            {
                                "compare_op": compare_op,
                                "field": all_urls_field,
                            },
                        )
                    else:
                        return compare_op
                elif operator in (PropertyOperator.ICONTAINS, PropertyOperator.NOT_ICONTAINS):
                    # For contains operators, delegate to _expr_to_compare_op which handles multiple values efficiently
                    if is_exception_string_array_property or is_visited_page_property:
                        # For exception properties and visited_page, use multiSearch optimization within arrayExists
                        multi_search_expr = _expr_to_compare_op(
                            ast.Field(chain=["v"]), value, operator, property, property.type != "session", team
                        )
                        if is_exception_string_array_property:
                            return parse_expr(
                                "arrayExists(v -> {expr}, {key})",
                                {"expr": multi_search_expr, "key": extracted_field},
                            )
                        else:  # is_visited_page_property
                            return parse_expr(
                                "arrayExists(v -> {expr}, {key})",
                                {"expr": multi_search_expr, "key": all_urls_field},
                            )
                    else:
                        return _expr_to_compare_op(expr, value, operator, property, property.type != "session", team)

                exprs = [
                    property_to_expr(
                        Property(
                            type=property.type,
                            key=property.key,
                            operator=property.operator,
                            group_type_index=property.group_type_index,
                            value=v,
                        ),
                        team,
                        scope,
                        strict=strict,
                    )
                    for v in value
                ]
                if (
                    operator == PropertyOperator.NOT_ICONTAINS
                    or operator == PropertyOperator.NOT_REGEX
                    or operator == PropertyOperator.IS_NOT
                    or operator == PropertyOperator.NOT_STARTS_WITH
                    or operator == PropertyOperator.NOT_ENDS_WITH
                ):
                    return ast.And(exprs=exprs)
                return ast.Or(exprs=exprs)

        expr = _expr_to_compare_op(
            expr=ast.Field(chain=["v"]) if (is_exception_string_array_property or is_visited_page_property) else expr,
            value=value,
            operator=operator,
            team=team,
            property=property,
            is_json_field=property.type != "session",
        )

        if is_exception_string_array_property:
            return parse_expr(
                "arrayExists(v -> {expr}, {key})",
                {"expr": expr, "key": extracted_field},
            )
        elif is_visited_page_property:
            # Handle IS_SET and IS_NOT_SET operators specially for arrays
            if operator == PropertyOperator.IS_SET:
                return ast.CompareOperation(
                    op=ast.CompareOperationOp.Gt,
                    left=ast.Call(name="length", args=[all_urls_field]),
                    right=ast.Constant(value=0),
                )
            elif operator == PropertyOperator.IS_NOT_SET:
                return ast.CompareOperation(
                    op=ast.CompareOperationOp.Eq,
                    left=ast.Call(name="length", args=[all_urls_field]),
                    right=ast.Constant(value=0),
                )
            else:
                return parse_expr(
                    "arrayExists(v -> {expr}, {key})",
                    {"expr": expr, "key": all_urls_field},
                )
        else:
            return expr
    elif property.type == "element":
        if scope == "person":
            raise NotImplementedError(f"property_to_expr for scope {scope} not implemented for type '{property.type}'")
        value = property.value
        operator = cast(Optional[PropertyOperator], property.operator) or PropertyOperator.EXACT
        if isinstance(value, list):
            if len(value) == 1:
                value = value[0]
            else:
                exprs = [
                    property_to_expr(
                        Property(
                            type=property.type,
                            key=property.key,
                            operator=property.operator,
                            group_type_index=property.group_type_index,
                            value=v,
                        ),
                        team,
                        scope,
                        strict=strict,
                    )
                    for v in value
                ]
                if (
                    operator == PropertyOperator.IS_NOT
                    or operator == PropertyOperator.NOT_ICONTAINS
                    or operator == PropertyOperator.NOT_REGEX
                    or operator == PropertyOperator.NOT_STARTS_WITH
                    or operator == PropertyOperator.NOT_ENDS_WITH
                ):
                    return ast.And(exprs=exprs)
                return ast.Or(exprs=exprs)

        if property.key == "selector" or property.key == "tag_name":
            if operator != PropertyOperator.EXACT and operator != PropertyOperator.IS_NOT:
                raise NotImplementedError(
                    f"property_to_expr for element {property.key} only supports exact and is_not operators, not {operator}"
                )
            expr = selector_to_expr(str(value)) if property.key == "selector" else tag_name_to_expr(str(value))
            if operator == PropertyOperator.IS_NOT:
                return ast.Call(name="not", args=[expr])
            return expr

        if property.key == "href":
            return _expr_to_compare_op(
                expr=ast.Field(chain=["elements_chain_href"]),
                value=value,
                operator=operator,
                team=team,
                property=property,
                is_json_field=False,
            )

        if property.key == "text":
            return parse_expr(
                "arrayExists(text -> {compare}, elements_chain_texts)",
                {
                    "compare": _expr_to_compare_op(
                        expr=ast.Field(chain=["text"]),
                        value=value,
                        operator=operator,
                        team=team,
                        property=property,
                        is_json_field=False,
                    )
                },
            )

        raise NotImplementedError(f"property_to_expr for type element not implemented for key {property.key}")
    elif property.type == "cohort" or property.type == "static-cohort" or property.type == "precalculated-cohort":
        if not team:
            raise Exception("Can not convert cohort property to expression without team")
        if not isinstance(property.value, (str, int)):
            raise ValidationError("Cohort property value must be a cohort ID")
        cohort = Cohort.objects.get(team__project_id=team.project_id, id=property.value)
        return ast.CompareOperation(
            left=ast.Field(chain=["id" if scope == "person" else "person_id"]),
            op=(
                ast.CompareOperationOp.NotInCohort
                # Kludge: negation is outdated but still used in places
                if property.negation or property.operator == PropertyOperator.NOT_IN.value
                else ast.CompareOperationOp.InCohort
            ),
            right=ast.Constant(value=cohort.pk),
        )

    # TODO: Add support for these types: "recording"

    raise NotImplementedError(
        f"property_to_expr not implemented for filter type {type(property).__name__} and {property.type}"
    )


def steps_to_expr(steps: list[ActionStepJSON], team: Team, events_alias: Optional[str] = None) -> ast.Expr:
    if len(steps) == 0:
        return ast.Constant(value=True)

    or_queries = []
    for step in steps:
        exprs: list[ast.Expr] = []
        if step.event:
            exprs.append(parse_expr("event = {event}", {"event": ast.Constant(value=step.event)}))

        if step.event == AUTOCAPTURE_EVENT:
            if step.selector:
                exprs.append(selector_to_expr(step.selector))
            if step.tag_name is not None:
                exprs.append(tag_name_to_expr(step.tag_name))
            if step.href is not None:
                if step.href_matching == "regex":
                    exprs.append(
                        ast.CompareOperation(
                            op=ast.CompareOperationOp.Regex,
                            left=ast.Field(chain=["elements_chain_href"]),
                            right=ast.Constant(value=step.href),
                        )
                    )
                elif step.href_matching == "contains":
                    exprs.append(
                        ast.CompareOperation(
                            op=ast.CompareOperationOp.ILike,
                            left=ast.Field(chain=["elements_chain_href"]),
                            right=ast.Constant(value=f"%{step.href}%"),
                        )
                    )
                else:
                    exprs.append(
                        ast.CompareOperation(
                            op=ast.CompareOperationOp.Eq,
                            left=ast.Field(chain=["elements_chain_href"]),
                            right=ast.Constant(value=step.href),
                        )
                    )
            if step.text is not None:
                value = step.text
                if step.text_matching == "regex":
                    exprs.append(
                        parse_expr(
                            "arrayExists(x -> x =~ {value}, elements_chain_texts)",
                            {"value": ast.Constant(value=value)},
                        )
                    )
                elif step.text_matching == "contains":
                    exprs.append(
                        parse_expr(
                            "arrayExists(x -> x ilike {value}, elements_chain_texts)",
                            {"value": ast.Constant(value=f"%{value}%")},
                        )
                    )
                else:
                    exprs.append(
                        parse_expr(
                            "arrayExists(x -> x = {value}, elements_chain_texts)",
                            {"value": ast.Constant(value=value)},
                        )
                    )

        if step.url:
            if step.url_matching == "exact":
                expr = ast.CompareOperation(
                    op=ast.CompareOperationOp.Eq,
                    left=ast.Field(
                        chain=(
                            [events_alias, "properties", "$current_url"]
                            if events_alias
                            else ["properties", "$current_url"]
                        )
                    ),
                    right=ast.Constant(value=step.url),
                )
            elif step.url_matching == "regex":
                expr = ast.CompareOperation(
                    op=ast.CompareOperationOp.Regex,
                    left=ast.Field(
                        chain=(
                            [events_alias, "properties", "$current_url"]
                            if events_alias
                            else ["properties", "$current_url"]
                        )
                    ),
                    right=ast.Constant(value=step.url),
                )
            else:
                expr = ast.CompareOperation(
                    op=ast.CompareOperationOp.Like,
                    left=ast.Field(
                        chain=(
                            [events_alias, "properties", "$current_url"]
                            if events_alias
                            else ["properties", "$current_url"]
                        )
                    ),
                    right=ast.Constant(value=f"%{step.url}%"),
                )
            exprs.append(expr)

        if step.properties:
            exprs.append(property_to_expr(step.properties, team))

        if len(exprs) == 1:
            or_queries.append(exprs[0])
        elif len(exprs) > 1:
            or_queries.append(ast.And(exprs=exprs))
        else:
            or_queries.append(ast.Constant(value=True))

    if len(or_queries) == 1:
        return or_queries[0]
    else:
        return ast.Or(exprs=or_queries)


def action_to_expr(action: Action, events_alias: Optional[str] = None) -> ast.Expr:
    return steps_to_expr(action.steps, action.team, events_alias)


def entity_to_expr(entity: RetentionEntity, team: Team) -> ast.Expr:
    if entity.type == TREND_FILTER_TYPE_ACTIONS and entity.id is not None:
        # action
        action_id = int(entity.id) if isinstance(entity.id, float) else entity.id
        try:
            action = Action.objects.get(pk=action_id, team=team)
        except Action.DoesNotExist:
            raise ValidationError(f"Action ID {entity.id} does not exist!")
        event_expr = action_to_expr(action)
    elif entity.id is None:
        # all events
        event_expr = ast.Constant(value=True)
    else:
        # event
        event_expr = parse_expr("events.event = {event}", {"event": ast.Constant(value=entity.id)})

    if entity.properties is not None and entity.properties != []:
        # add property filters
        filter_expr = property_to_expr(entity.properties, team)
        return ast.And(exprs=[event_expr, filter_expr])
    else:
        return event_expr


def tag_name_to_expr(tag_name: str):
    regex = rf"(^|;){tag_name}(\.|$|;|:)"
    expr = parse_expr("elements_chain =~ {regex}", {"regex": ast.Constant(value=str(regex))})
    return expr


def selector_to_expr(selector_string: str):
    selector = Selector(selector_string, escape_slashes=False)
    exprs = []
    regex = build_selector_regex(selector)
    exprs.append(parse_expr("elements_chain =~ {regex}", {"regex": ast.Constant(value=regex)}))

    useful_elements: list[ast.Expr] = []
    for part in selector.parts:
        if "tag_name" in part.data:
            if part.data["tag_name"] in Element.USEFUL_ELEMENTS:
                useful_elements.append(ast.Constant(value=part.data["tag_name"]))

        if "attr_id" in part.data:
            id_expr = parse_expr(
                "indexOf(elements_chain_ids, {value}) > 0", {"value": ast.Constant(value=part.data["attr_id"])}
            )
            if len(selector.parts) == 1 and len(part.data.keys()) == 1:
                # OPTIMIZATION: if there's only one selector part and that only filters on an ID, we don't need to also query elements_chain separately
                return id_expr
            exprs.append(id_expr)
    if len(useful_elements) > 0:
        exprs.append(
            parse_expr(
                "arrayCount(x -> x IN {value}, elements_chain_elements) > 0",
                {"value": ast.Array(exprs=useful_elements)},
            )
        )

    if len(exprs) == 1:
        return exprs[0]
    return ast.And(exprs=exprs)


def get_property_type(property):
    return get_from_dict_or_attr(property, "type")


def get_property_key(property):
    return get_from_dict_or_attr(property, "key")


def get_property_value(property):
    return get_from_dict_or_attr(property, "value")


def get_property_operator(property):
    return get_from_dict_or_attr(property, "operator")


def get_lowercase_index_hint(property, team: Team) -> ast.Call:
    """
    Returns an index hint for a case insensitive index on `lower(key)`
    e.g. for the property `body ILIKE '%STR%'` return `indexHint(lower(body) ILIKE '%str%')`
         this means we can use ngram indexes on `lower(body)` efficiently
    """
    expr = property_to_expr(property, team=team)
    return ast.Call(name="indexHint", args=[_LowercaseIndexRewriter().visit(expr)])


class _LowercaseIndexRewriter(CloningVisitor):
    """Rewrites an expression tree so it can leverage a case-insensitive index on ``lower(key)``.

    Transformations applied:
    - All ``toString(x)`` calls are unwrapped to just ``x`` (stripped, not replaced).
    - ``ILike`` → ``lower(left) Like lower_const`` / ``NotILike`` → ``lower(left) NotLike lower_const``
    - ``multiSearchAnyCaseInsensitive(haystack, needles)`` → ``multiSearchAny(lower(haystack), lowered_needles)``
    """

    def visit_call(self, node: ast.Call):
        if node.name == "toString" and len(node.args) == 1:
            # Strip toString
            return self.visit(node.args[0])

        if node.name == "ifNull" and len(node.args) >= 1:
            # Strip ifNull
            return self.visit(node.args[0])

        if node.name == "multiSearchAnyCaseInsensitive" and len(node.args) == 2:
            # multiSearchAnyCaseInsensitive(haystack, needles)
            # → multiSearchAny(lower(haystack), lowered_needles)
            haystack = self.visit(node.args[0])
            haystack = ast.Call(name="lower", args=[haystack])
            needles = node.args[1]
            # Lowercase all needle constants
            if isinstance(needles, ast.Array):
                needles = ast.Array(
                    exprs=[
                        ast.Constant(value=str(e.value).lower())
                        if isinstance(e, ast.Constant) and isinstance(e.value, str)
                        else self.visit(e)
                        for e in needles.exprs
                    ]
                )
            else:
                needles = self.visit(needles)
            return ast.Call(name="multiSearchAny", args=[haystack, needles])

        return super().visit_call(node)

    def visit_compare_operation(self, node: ast.CompareOperation):
        op = node.op
        wrap_left_lower = False

        if op == ast.CompareOperationOp.ILike:
            op = ast.CompareOperationOp.Like
            wrap_left_lower = True
        elif op == ast.CompareOperationOp.NotILike:
            op = ast.CompareOperationOp.NotLike
            wrap_left_lower = True

        left = self.visit(node.left)
        right = self.visit(node.right)

        if wrap_left_lower:
            left = ast.Call(name="lower", args=[left])
            if isinstance(right, ast.Constant) and isinstance(right.value, str):
                right = ast.Constant(value=right.value.lower())

        return ast.CompareOperation(
            left=left,
            right=right,
            op=op,
        )


def operator_is_negative(operator: PropertyOperator) -> bool:
    return operator in [
        PropertyOperator.IS_NOT,
        PropertyOperator.NOT_ICONTAINS,
        PropertyOperator.NOT_ICONTAINS_MULTI,
        PropertyOperator.NOT_STARTS_WITH,
        PropertyOperator.NOT_ENDS_WITH,
        PropertyOperator.NOT_REGEX,
        PropertyOperator.IS_NOT_SET,
        PropertyOperator.NOT_BETWEEN,
        PropertyOperator.NOT_IN,
    ]