insights/products/alerts/backend/api/alert.py

import uuid
from typing import Annotated, Any, cast
from zoneinfo import ZoneInfo

from django.db import transaction
from django.db.models import OuterRef, Q, QuerySet, Subquery

import hanzo_insights
from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema, extend_schema_view
from pydantic import (
    Field as PydanticField,
    RootModel,
)
from rest_framework import serializers, status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied, ValidationError
from rest_framework.response import Response

from insights.schema import (
    AlertCalculationInterval,
    AlertCondition,
    DetectorConfig,
    FunnelsAlertConfig,
    InsightsQLAlertConfig,
    InsightThreshold,
    MetricsAlertConfig,
    NodeKind,
    TrendsAlertConfig,
)

from insights.api.documentation import extend_schema_field
from insights.api.routing import TeamAndOrgViewSetMixin
from insights.api.scoped_related_fields import TeamScopedPrimaryKeyRelatedField
from insights.api.shared import SearchMatchTypeSerializerMixin, UserBasicSerializer
from insights.event_usage import get_request_analytics_properties
from insights.exceptions_capture import capture_exception
from insights.helpers.trigram_search import (
    MAX_SEARCH_LENGTH,
    NAME_FIELD,
    apply_trigram_search,
    drop_similar_when_exact_exists,
)
from insights.models import User
from insights.permissions import get_authenticator_scopes
from insights.rate_limit import AlertTestDeliveryThrottle
from insights.resource_limits import LimitKey, check_count_limit
from insights.schema_migrations.upgrade_manager import upgrade_query
from insights.tasks.alerts.detector import MAX_DETECTOR_BREAKDOWN_VALUES
from insights.tasks.alerts.schedule_restriction import validate_and_normalize_schedule_restriction
from insights.tasks.alerts.utils import (
    next_check_at_after_schedule_restriction_change,
    send_test_alert_email,
    trigger_alert_insights_functions,
)
from insights.utils import relative_date_parse

from products.alerts.backend.api.alert_schedule_restriction import AlertScheduleRestriction
from products.alerts.backend.destinations import count_active_alert_destinations
from products.alerts.backend.evaluation.contract import AlertExtractionError
from products.alerts.backend.evaluation.detector import simulate_detector_on_insight
from products.alerts.backend.evaluation.validation import (
    THRESHOLD_BOUNDS_REQUIRED_MESSAGE,
    should_default_check_ongoing_interval,
    validate_alert_config,
)
from products.alerts.backend.insight_alert_state_machine import (
    apply_disable,
    apply_enable,
    apply_snooze,
    apply_threshold_change,
    apply_unsnooze,
)
from products.alerts.backend.models.alert import AlertCheck, AlertConfiguration, AlertSubscription, Threshold
from products.product_analytics.backend.models.insight import Insight

INSIGHT_ALERT_FIRING_EVENT = "$insight_alert_firing"


def _validate_interval_entitlement(
    *,
    calculation_interval: str | AlertCalculationInterval | None,
    organization,
) -> None:
    if error := AlertConfiguration.interval_entitlement_error(
        calculation_interval=calculation_interval,
        organization=organization,
    ):
        raise ValidationError({"calculation_interval": [error]})


def _require_insight_viewer_access(context: dict[str, Any], insight: Insight) -> None:
    # Team scoping alone doesn't gate per-object insight access controls, so require viewer access
    # explicitly — alert write/simulate access must not expose a restricted insight's results.
    if not context["view"].user_access_control.check_access_level_for_object(insight, "viewer"):
        raise ValidationError("Viewer access to this insight is required.")


@extend_schema_field(InsightThreshold)  # type: ignore[arg-type]
class ThresholdConfigurationField(serializers.JSONField):
    pass


@extend_schema_field(AlertCondition)  # type: ignore[arg-type]
class AlertConditionField(serializers.JSONField):
    pass


def _insight_alert_flag_enabled(context: dict[str, Any], flag: str) -> bool:
    # Scope the flag to the alert's organization (via team scope), not the user's current
    # organization — otherwise a user in multiple orgs could flip their current org to a
    # flag-on org and create an alert in a team where the flag is disabled. get_organization is
    # always injected by TeamAndOrgViewSetMixin; access it unconditionally so the org-scoping
    # invariant can't silently degrade to an unscoped check.
    user = context["request"].user
    org = context["get_organization"]()
    return bool(
        hanzo_insights.feature_enabled(
            flag,
            str(user.distinct_id),
            groups={"organization": str(org.id)},
        )
    )


def _enforce_alert_feature_flags(context: dict[str, Any], insight: Insight) -> None:
    # Shared by create/update (AlertSerializer) and simulate (AlertSimulateSerializer), so a
    # flag-gated insight kind gets the same rejection on every alerts entry point.
    kind = insight.alertable_query_kind
    if kind is None:
        raise ValidationError("Alerts are not supported for this insight.")
    # Gated on the Metrics product flag itself: anyone who can see the product can alert on it.
    if kind == NodeKind.METRICS_QUERY and not _insight_alert_flag_enabled(context, "metrics"):
        raise ValidationError("Metrics insight alerts are not enabled for your account.")


# Matches insights whose (bare or one-level-wrapped) query is a MetricsQuery — the shapes the
# product persists today. Used to hide metric alert data from tokens lacking the metrics scope.
_METRICS_INSIGHT_QUERY_FILTER = Q(insight__query__kind="MetricsQuery") | Q(insight__query__source__kind="MetricsQuery")


def _token_lacks_metrics_scope(request) -> bool:
    # Session auth (no token scopes) is exempt: those users are gated by team membership and
    # insight viewer access instead.
    key_scopes = get_authenticator_scopes(request.successful_authenticator)
    if key_scopes is None or "*" in key_scopes:
        return False
    return not any(scope in key_scopes for scope in ("metrics:read", "metrics:write"))


def _require_metrics_scope_for_programmatic_auth(context: dict[str, Any], insight: Insight) -> None:
    # An alert on a metrics insight executes the query as `created_by` and delivers the computed
    # value and labels in notifications, so a programmatic token must also carry the metrics data
    # scope — otherwise `alert:write` alone becomes a metrics-read oracle bypassing the explicit
    # product-scope gate on the query endpoints.
    if insight.alertable_query_kind != NodeKind.METRICS_QUERY:
        return
    if _token_lacks_metrics_scope(context["request"]):
        raise PermissionDenied("API key missing required scope 'metrics:read'")


class AlertConfigUnion(RootModel):
    """Per-insight-kind alert config, discriminated by ``type`` — keeps the OpenAPI (and the
    generated frontend types and MCP tool schemas) in sync with every kind alerts support."""

    root: Annotated[
        TrendsAlertConfig | InsightsQLAlertConfig | FunnelsAlertConfig | MetricsAlertConfig,
        PydanticField(discriminator="type"),
    ]


@extend_schema_field(AlertConfigUnion)  # type: ignore[arg-type]
class AlertConfigField(serializers.JSONField):
    pass


@extend_schema_field(DetectorConfig)  # type: ignore[arg-type]
class DetectorConfigField(serializers.JSONField):
    pass


@extend_schema_field(AlertScheduleRestriction)  # type: ignore[arg-type]
class ScheduleRestrictionField(serializers.JSONField):
    pass


class ThresholdSerializer(serializers.ModelSerializer):
    configuration = ThresholdConfigurationField(
        help_text="Threshold bounds and type. Includes bounds (lower/upper floats) and type (absolute or percentage). "
        "For threshold-based alerts (no detector_config), at least one of lower or upper must be set.",
    )
    name = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=255,
        help_text="Optional name for the threshold.",
    )

    class Meta:
        model = Threshold
        fields = [
            "id",
            "created_at",
            "name",
            "configuration",
        ]
        read_only_fields = [
            "id",
            "created_at",
        ]

    def validate(self, data):
        instance = Threshold(**data)
        instance.clean()
        return data


class AlertCheckSerializer(serializers.ModelSerializer):
    targets_notified = serializers.SerializerMethodField()
    investigation_notebook_short_id = serializers.SerializerMethodField(
        help_text="Short ID of the Notebook produced by the investigation agent, when the agent ran for this check."
    )

    class Meta:
        model = AlertCheck
        fields = [
            "id",
            "created_at",
            "calculated_value",
            "state",
            "targets_notified",
            "anomaly_scores",
            "triggered_points",
            "triggered_dates",
            "interval",
            "triggered_metadata",
            "investigation_status",
            "investigation_verdict",
            "investigation_summary",
            "investigation_notebook_short_id",
            "notification_sent_at",
            "notification_suppressed_by_agent",
        ]
        read_only_fields = fields

    def get_targets_notified(self, instance: AlertCheck) -> bool:
        return instance.targets_notified != {}

    def get_investigation_notebook_short_id(self, instance: AlertCheck) -> str | None:
        notebook = instance.investigation_notebook
        return notebook.short_id if notebook is not None else None


class AlertSubscriptionSerializer(serializers.ModelSerializer):
    # nosemgrep: unscoped-primary-key-related-field — User model is not team-scoped; validate() checks team membership
    user = serializers.PrimaryKeyRelatedField(queryset=User.objects.filter(is_active=True), required=True)

    class Meta:
        model = AlertSubscription
        fields = ["id", "user", "alert_configuration"]
        read_only_fields = ["id", "alert_configuration"]

    def validate(self, data):
        user: User = data["user"]
        alert_configuration = data["alert_configuration"]

        if not user.teams.filter(pk=alert_configuration.team_id).exists():
            raise serializers.ValidationError("User does not belong to the same organization as the alert's project.")

        return data


@extend_schema_field(
    {
        "type": "string",
        "nullable": True,
        "description": "Snooze the alert until this time. Pass a relative date string (e.g. '2h', '1d') or null to unsnooze.",
    }
)
class RelativeDateTimeField(serializers.DateTimeField):
    def to_internal_value(self, data):
        return data


class AlertSerializer(SearchMatchTypeSerializerMixin, serializers.ModelSerializer):
    created_by = UserBasicSerializer(read_only=True)
    checks = AlertCheckSerializer(
        many=True,
        read_only=True,
        help_text="Alert check results. By default returns the last 5. Use checks_date_from and checks_date_to (e.g. '-24h', '-7d') to get checks within a time window, checks_limit to cap how many are returned (default 5, max 500), and checks_offset to skip the newest N checks for pagination (0-based). Newest checks first. Only populated on retrieve.",
    )
    checks_total = serializers.SerializerMethodField(
        read_only=True,
        help_text="Total alert checks matching the retrieve filters (date window). Only set on alert retrieve; omitted otherwise.",
    )
    threshold = ThresholdSerializer(
        help_text="Threshold configuration with bounds and type for evaluating the alert.",
    )
    condition = AlertConditionField(
        required=False,
        allow_null=True,
        help_text="Alert condition type. Determines how the value is evaluated: absolute_value, relative_increase, or relative_decrease.",
    )
    config = AlertConfigField(
        required=False,
        allow_null=True,
        help_text=(
            "Per-insight-kind alert configuration, discriminated by `type`. TrendsAlertConfig: series_index "
            "(which series to monitor) and check_ongoing_interval (whether to check the current incomplete "
            "interval). InsightsQLAlertConfig (SQL insights): column (which result column to evaluate, defaults to "
            "the single numeric column), evaluation ('last_row' checks the latest value of an oldest->newest query, "
            "'first_row' checks the first value of a newest->oldest query, 'any_row' fires if any row breaches), and "
            "label_column (names the evaluated row(s) in breach messages, in every evaluation mode). "
            "FunnelsAlertConfig (funnel insights): funnel_step (the step to monitor, null for the overall "
            "last step), metric ('conversion_from_start' or 'conversion_from_previous'), and "
            "check_ongoing_interval (historical-trend funnels: also evaluate the current in-progress period). "
            "Steps funnels support only absolute_value conditions; historical-trend funnels also support "
            "relative_increase/relative_decrease (compared against the prior period)."
        ),
    )
    detector_config = DetectorConfigField(required=False, allow_null=True)
    insight = TeamScopedPrimaryKeyRelatedField(
        queryset=Insight.objects.all(),
        help_text="Insight ID monitored by this alert. Note: Response returns full InsightBasicSerializer object.",
    )
    name = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=255,
        help_text="Human-readable name for the alert.",
    )
    # nosemgrep: unscoped-primary-key-related-field — User model is not team-scoped; validate_subscribed_users() checks team membership
    subscribed_users = serializers.PrimaryKeyRelatedField(
        queryset=User.objects.filter(is_active=True),
        many=True,
        required=True,
        allow_empty=True,
        help_text="User IDs to subscribe to this alert. Note: Response returns full UserBasicSerializer object.",
    )
    enabled = serializers.BooleanField(
        required=False,
        help_text="Whether the alert is actively being evaluated.",
    )
    calculation_interval = serializers.ChoiceField(
        choices=AlertConfiguration.CALCULATION_INTERVAL_CHOICES,
        required=False,
        help_text="How often the alert is checked: real time (Scale+), every 15 minutes (Boost+), hourly, daily, weekly, or monthly.",
    )
    snoozed_until = RelativeDateTimeField(
        allow_null=True,
        required=False,
        help_text="Snooze the alert until this time. Pass a relative date string (e.g. '2h', '1d') or null to unsnooze.",
    )
    skip_weekend = serializers.BooleanField(
        required=False,
        allow_null=True,
        help_text="Skip alert evaluation on weekends (Saturday and Sunday, local to project timezone).",
    )
    schedule_restriction = ScheduleRestrictionField(
        required=False,
        allow_null=True,
        help_text="Blocked local time windows (HH:MM in the project timezone). Interval is half-open [start, end): "
        "start inclusive, end exclusive. Use blocked_windows array of {start, end}. Null disables.",
    )
    investigation_agent_enabled = serializers.BooleanField(
        required=False,
        help_text="When enabled, an investigation agent runs on the state transition to firing and writes findings to a Notebook linked from the alert check. Only effective for detector-based (anomaly) alerts.",
    )
    investigation_gates_notifications = serializers.BooleanField(
        required=False,
        help_text="When enabled (and investigation_agent_enabled is on), notification dispatch is held until the investigation agent produces a verdict. Notifications are suppressed when the verdict is false_positive (and optionally when inconclusive). A safety-net task force-fires after a few minutes if the investigation stalls.",
    )
    investigation_inconclusive_action = serializers.ChoiceField(
        choices=[("notify", "Notify"), ("suppress", "Suppress")],
        required=False,
        help_text="How to handle an 'inconclusive' verdict: whether gated notifications fire and whether the investigation surfaces in the Signals inbox. 'notify' is the safe default — an agent that can't be sure is itself useful signal. False positives never reach the inbox regardless of this setting.",
    )
    state = serializers.CharField(
        read_only=True,
        help_text="Current alert state: Firing, Not firing, Errored, or Snoozed.",
    )
    last_value = serializers.FloatField(
        read_only=True,
        allow_null=True,
        help_text="The last calculated value from the most recent alert check.",
    )

    def get_checks_total(self, obj: AlertConfiguration) -> int | None:
        return getattr(obj, "checks_total", None)

    class Meta:
        model = AlertConfiguration
        fields = [
            "id",
            "created_by",
            "created_at",
            "insight",
            "name",
            "subscribed_users",
            "threshold",
            "condition",
            "state",
            "enabled",
            "last_notified_at",
            "last_checked_at",
            "next_check_at",
            "checks",
            "checks_total",
            "config",
            "detector_config",
            "calculation_interval",
            "snoozed_until",
            "skip_weekend",
            "schedule_restriction",
            "last_value",
            "investigation_agent_enabled",
            "investigation_gates_notifications",
            "investigation_inconclusive_action",
            "search_match_type",
        ]
        read_only_fields = [
            "id",
            "created_at",
            "state",
            "last_notified_at",
            "last_checked_at",
            "next_check_at",
        ]

    def to_representation(self, instance):
        # Deferred: the insight API serializer drags the product-analytics query-runner chain;
        # keeping it out of module scope lets this module connect its delete receivers at
        # AppConfig.ready() without pulling that chain onto startup.
        from products.product_analytics.backend.api.insight import InsightBasicSerializer  # noqa: PLC0415

        data = super().to_representation(instance)
        data["subscribed_users"] = UserBasicSerializer(instance.subscribed_users.all(), many=True, read_only=True).data
        data["insight"] = InsightBasicSerializer(instance.insight).data
        if data.get("checks_total") is None:
            data.pop("checks_total", None)
        return data

    def add_threshold(self, threshold_data, validated_data):
        threshold_instance = Threshold.objects.create(
            **threshold_data,
            team_id=self.context["team_id"],
            created_by=self.context["request"].user,
            insight_id=validated_data["insight"].id,
        )
        return threshold_instance

    def create(self, validated_data: dict) -> AlertConfiguration:
        validated_data["team_id"] = self.context["team_id"]
        validated_data["created_by"] = self.context["request"].user
        team = self.context["get_team"]()
        current_count = AlertConfiguration.objects.filter(team_id=team.id).count()
        check_count_limit(
            team=team,
            key=LimitKey.MAX_ALERTS_PER_TEAM,
            current_count=current_count,
            user=self.context["request"].user,
        )
        subscribed_users = validated_data.pop("subscribed_users")
        threshold_data = validated_data.pop("threshold", None)

        if threshold_data:
            threshold_instance = self.add_threshold(threshold_data, validated_data)
            validated_data["threshold"] = threshold_instance

        instance: AlertConfiguration = super().create(validated_data)

        for user in subscribed_users:
            AlertSubscription.objects.create(
                user=user, alert_configuration=instance, created_by=self.context["request"].user
            )

        instance.report_created(
            self.context["request"].user,
            analytics_props=get_request_analytics_properties(self.context["request"]),
        )
        return instance

    @transaction.atomic
    def update(self, instance, validated_data):
        enabled_changed = "enabled" in validated_data and validated_data["enabled"] != instance.enabled
        resulting_enabled = validated_data.get("enabled", instance.enabled)
        if enabled_changed and validated_data["enabled"]:
            apply_enable(instance)

        snoozed_until_param = validated_data.pop("snoozed_until", serializers.empty)
        snooze_changed = snoozed_until_param is not serializers.empty
        snoozed_until = None
        if snooze_changed and snoozed_until_param is not None:
            snoozed_until = relative_date_parse(
                snoozed_until_param, ZoneInfo("UTC"), increase=True, always_truncate=True
            )

        conditions_or_threshold_changed = False

        threshold_data = validated_data.pop("threshold", None)
        if threshold_data is not None:
            if threshold_data == {}:
                instance.threshold = None
                conditions_or_threshold_changed = True
            elif instance.threshold:
                previous_threshold_configuration = instance.threshold.configuration
                threshold_instance = instance.threshold
                for key, value in threshold_data.items():
                    setattr(threshold_instance, key, value)
                threshold_instance.save()
                if previous_threshold_configuration != threshold_instance.configuration:
                    conditions_or_threshold_changed = True
            else:
                threshold_instance = self.add_threshold(threshold_data, validated_data)
                validated_data["threshold"] = threshold_instance
                conditions_or_threshold_changed = True

        subscribed_users = validated_data.pop("subscribed_users", None)
        if subscribed_users is not None:
            AlertSubscription.objects.filter(alert_configuration=instance).exclude(user__in=subscribed_users).delete()
            for user in subscribed_users:
                # nosemgrep: idor-lookup-without-team (user team membership validated by viewset)
                AlertSubscription.objects.get_or_create(
                    user=user, alert_configuration=instance, defaults={"created_by": self.context["request"].user}
                )

        calculation_interval_changed = (
            "calculation_interval" in validated_data
            and validated_data["calculation_interval"] != instance.calculation_interval
        )
        if conditions_or_threshold_changed or calculation_interval_changed:
            if conditions_or_threshold_changed:
                apply_threshold_change(instance)
            instance.next_check_at = None

        if snooze_changed:
            instance.snoozed_until = snoozed_until
            if snoozed_until_param is None:
                apply_unsnooze(instance)
            else:
                apply_snooze(instance)

        if not resulting_enabled:
            apply_disable(instance)

        if snooze_changed:
            AlertCheck.objects.create(
                alert_configuration=instance,
                calculated_value=None,
                condition=instance.condition,
                targets_notified={},
                state=instance.state,
                error=None,
            )

        schedule_restriction_changed = False
        if "schedule_restriction" in validated_data:
            new_sr = validated_data["schedule_restriction"]
            if new_sr != instance.schedule_restriction:
                schedule_restriction_changed = True

        instance = super().update(instance, validated_data)
        if schedule_restriction_changed:
            instance.next_check_at = next_check_at_after_schedule_restriction_change(instance)
            instance.save(update_fields=["next_check_at"])

        instance.report_updated(
            self.context["request"].user,
            analytics_props=get_request_analytics_properties(self.context["request"]),
        )
        return instance

    def validate_detector_config(self, value):
        if value is None:
            return value

        import pydantic

        try:
            validated = DetectorConfig.model_validate(value)
        except pydantic.ValidationError:
            raise ValidationError("Invalid detector configuration.")

        # Ensemble requires at least 2 sub-detectors
        root = validated.root if hasattr(validated, "root") else validated
        if getattr(root, "type", None) == "ensemble" and hasattr(root, "detectors"):
            if len(root.detectors) < 2:
                raise ValidationError("Ensemble detector requires at least 2 sub-detectors.")
            for sub in root.detectors:
                sub_dict: dict = sub.model_dump() if hasattr(sub, "model_dump") else sub  # type: ignore[assignment]
                self._validate_detector_params(sub_dict)
        else:
            self._validate_detector_params(value)

        return validated.model_dump() if hasattr(validated, "model_dump") else value

    @staticmethod
    def _validate_detector_params(config: dict) -> None:
        """Validate detector parameter ranges match frontend constraints."""
        # Parameter ranges: (min, max, name)
        PARAM_RANGES: dict[str, tuple[float, float, str]] = {
            "threshold": (0.0, 1.0, "Sensitivity threshold"),
            "window": (5, 1000, "Window size"),
            "n_estimators": (10, 500, "Number of trees"),
            "n_neighbors": (1, 50, "Number of neighbors"),
            "n_bins": (5, 50, "Number of bins"),
            "multiplier": (0.5, 10.0, "IQR multiplier"),
            "training_offset_n": (1, 500, "Training offset"),
        }

        for param, (min_val, max_val, label) in PARAM_RANGES.items():
            val = config.get(param)
            if val is not None:
                if val < min_val or val > max_val:
                    raise ValidationError(f"{label} must be between {min_val} and {max_val}.")

        preprocessing = config.get("preprocessing")
        if preprocessing and isinstance(preprocessing, dict):
            smooth_n = preprocessing.get("smooth_n")
            if smooth_n is not None and (smooth_n < 0 or smooth_n > 30):
                raise ValidationError("Smoothing window must be between 0 and 30.")
            lags_n = preprocessing.get("lags_n")
            if lags_n is not None and (lags_n < 0 or lags_n > 10):
                raise ValidationError("Lag features must be between 0 and 10.")

    def validate_snoozed_until(self, value):
        if value is not None and not isinstance(value, str):
            raise ValidationError("snoozed_until has to be passed in string format")

        return value

    def validate_insight(self, value):
        if not value:
            return value
        _require_insight_viewer_access(self.context, value)
        _enforce_alert_feature_flags(self.context, value)
        return value

    def validate_subscribed_users(self, value):
        for user in value:
            if not user.teams.filter(pk=self.context["team_id"]).exists():
                raise ValidationError("User does not belong to the same organization as the alert's project.")
        return value

    def validate_schedule_restriction(self, value):
        try:
            return validate_and_normalize_schedule_restriction(value)
        except ValueError:
            raise serializers.ValidationError("Invalid schedule restriction.")

    def validate(self, attrs):
        if attrs.get("insight") and attrs["insight"].team.id != self.context["team_id"]:
            raise ValidationError({"insight": ["This insight does not belong to your project."]})

        condition = attrs.get("condition", self.instance.condition if self.instance else None)
        config = attrs.get("config", self.instance.config if self.instance else None)
        insight = attrs.get("insight") or (self.instance.insight if self.instance else None)
        if insight is None:
            raise ValidationError({"insight": ["Insight is required."]})
        # Enforced from the object-level validate() so it runs on every create and update — including
        # a PATCH that omits `insight` (which skips the field-level validator), so an alert can't be
        # repointed at a flag-gated insight kind (or a scope-gated one) in an account where the flag
        # is off or the token lacks the data scope. The model stays flag-agnostic so existing alerts
        # keep working when the flag is off.
        _enforce_alert_feature_flags(self.context, insight)
        _require_metrics_scope_for_programmatic_auth(self.context, insight)
        with upgrade_query(insight):
            query = insight.query
            if query is None:
                raise ValidationError({"insight": ["Insight has no valid query."]})

        threshold_config = None
        if "threshold" in attrs and isinstance(attrs["threshold"], dict):
            threshold_config = attrs["threshold"].get("configuration")
        elif self.instance and self.instance.threshold:
            threshold_config = self.instance.threshold.configuration

        calculation_interval = attrs.get(
            "calculation_interval",
            self.instance.calculation_interval if self.instance else AlertCalculationInterval.DAILY,
        )

        if "detector_config" in attrs:
            detector_config = attrs["detector_config"]
        elif self.instance is not None:
            detector_config = self.instance.detector_config
        else:
            detector_config = None

        require_threshold_bounds = detector_config is None and (
            self.instance is None or "threshold" in attrs or "detector_config" in attrs
        )

        # Mirror the UI's default for cadences finer than the insight interval. Applied before
        # validate_alert_config so the validated config is the persisted config.
        if isinstance(config, dict) and config.get("check_ongoing_interval") is None:
            if should_default_check_ongoing_interval(
                query=query,
                config=config,
                condition=condition,
                threshold_config=threshold_config,
                calculation_interval=calculation_interval,
            ):
                config = {**config, "check_ongoing_interval": True}
                attrs["config"] = config

        try:
            validate_alert_config(
                query,
                condition,
                config,
                threshold_config,
                calculation_interval,
                detector_config=detector_config,
                require_threshold_bounds=require_threshold_bounds,
            )
        except ValueError as e:
            if str(e) == THRESHOLD_BOUNDS_REQUIRED_MESSAGE:
                raise ValidationError({"threshold": {"configuration": [THRESHOLD_BOUNDS_REQUIRED_MESSAGE]}})
            raise ValidationError(str(e))

        organization = self.context["get_organization"]()
        _validate_interval_entitlement(
            calculation_interval=calculation_interval,
            organization=organization,
        )

        # Investigation agent is supported for detector-based alerts (the agent
        # workflow) and for metrics threshold alerts (a synchronous facade
        # investigation attached to the firing check) — nothing else.
        investigation_enabled = attrs.get(
            "investigation_agent_enabled",
            self.instance.investigation_agent_enabled if self.instance else False,
        )
        if investigation_enabled:
            detector_config = attrs.get(
                "detector_config",
                self.instance.detector_config if self.instance else None,
            )
            if not detector_config and insight.alertable_query_kind != NodeKind.METRICS_QUERY:
                raise ValidationError(
                    {
                        "investigation_agent_enabled": [
                            "Investigation agent is only supported for anomaly detection alerts."
                        ]
                    }
                )

        # Notification gating only makes sense when the investigation agent is on —
        # otherwise there's no verdict to wait for and the safety-net task would
        # end up being the only notifier, which defeats the feature.
        gates_notifications = attrs.get(
            "investigation_gates_notifications",
            self.instance.investigation_gates_notifications if self.instance else False,
        )
        if gates_notifications and not investigation_enabled:
            raise ValidationError(
                {
                    "investigation_gates_notifications": [
                        "Notification gating requires investigation_agent_enabled=true."
                    ]
                }
            )

        if self.context["request"].method == "POST":
            if msg := AlertConfiguration.check_alert_limit(self.context["team_id"], self.context["get_organization"]()):
                raise ValidationError({"alert": [msg]})

        existing_enabled = self.instance.enabled if self.instance else True
        if msg := AlertConfiguration.real_time_alert_validation_error(
            team_id=self.context["team_id"],
            organization=organization,
            calculation_interval=calculation_interval,
            enabled=attrs.get("enabled", existing_enabled) is True,
            existing=self.instance,
        ):
            hanzo_insights.capture(
                distinct_id=str(self.context["request"].user.distinct_id),
                event="real time alert limit reached",
                properties={
                    "team_id": self.context["team_id"],
                    "organization_id": str(organization.id),
                },
                groups={"organization": str(organization.id)},
            )
            raise ValidationError({"calculation_interval": [msg]})

        return attrs


class AlertSimulateSerializer(serializers.Serializer):
    insight = TeamScopedPrimaryKeyRelatedField(
        queryset=Insight.objects.all(),
        help_text="Insight ID to simulate the detector on.",
    )
    detector_config = DetectorConfigField(
        help_text="Detector configuration to simulate.",
    )
    # TODO: fold series_index and date_from into a per-kind range on `config` once a second insight
    # kind needs a range knob. They stay flat today because date_from is a preview-only range with
    # no home in the persisted alert config, and trends is the only kind with a range knob — but the
    # duplication (series_index already lives in TrendsAlertConfig) should be removed at that point.
    series_index = serializers.IntegerField(
        default=0,
        help_text="Zero-based index of the series to analyze (trends insights only).",
    )
    date_from = serializers.CharField(
        required=False,
        allow_null=True,
        default=None,
        help_text="Relative date string for how far back to simulate (e.g. '-24h', '-30d', '-4w'). "
        "If not provided, uses the detector's minimum required samples. Trends insights only — a SQL "
        "query's own rows are the series.",
    )
    config = AlertConfigField(
        required=False,
        allow_null=True,
        default=None,
        help_text="Per-insight-kind alert config. For SQL insights, selects the evaluated column and "
        "read direction (last_row/first_row) so the preview matches the alert; ignored for trends.",
    )

    def validate_insight(self, value):
        _require_insight_viewer_access(self.context, value)
        # Same feature gate as create/update: a flag-gated insight kind must get the gated
        # rejection here too, not fall through to the unsupported-detector error.
        _enforce_alert_feature_flags(self.context, value)
        return value

    def validate_detector_config(self, value):
        if value is None:
            raise ValidationError("detector_config is required.")

        import pydantic

        try:
            validated = DetectorConfig.model_validate(value)
        except pydantic.ValidationError:
            raise ValidationError("Invalid detector configuration.")

        root = validated.root if hasattr(validated, "root") else validated
        if getattr(root, "type", None) == "ensemble" and hasattr(root, "detectors"):
            if len(root.detectors) < 2:
                raise ValidationError("Ensemble detector requires at least 2 sub-detectors.")
            for sub in root.detectors:
                sub_dict: dict = sub.model_dump() if hasattr(sub, "model_dump") else sub  # type: ignore[assignment]
                AlertSerializer._validate_detector_params(sub_dict)
        else:
            AlertSerializer._validate_detector_params(value)

        return validated.model_dump() if hasattr(validated, "model_dump") else value


class BreakdownSimulationResultSerializer(serializers.Serializer):
    label = serializers.CharField(help_text="Breakdown value label.")  # type: ignore[assignment]
    data = serializers.ListField(child=serializers.FloatField(), help_text="Data values for each point.")  # type: ignore[assignment]
    dates = serializers.ListField(child=serializers.CharField(), help_text="Date labels for each point.")
    scores = serializers.ListField(
        child=serializers.FloatField(allow_null=True),
        help_text="Anomaly score for each point.",
    )
    triggered_indices = serializers.ListField(
        child=serializers.IntegerField(), help_text="Indices of points flagged as anomalies."
    )
    triggered_dates = serializers.ListField(
        child=serializers.CharField(), help_text="Dates of points flagged as anomalies."
    )
    total_points = serializers.IntegerField(help_text="Total number of data points analyzed.")
    anomaly_count = serializers.IntegerField(help_text="Number of anomalies detected.")
    sub_detector_scores = serializers.ListField(
        child=serializers.DictField(),
        required=False,
        help_text="Per-sub-detector scores for ensemble detectors.",
    )


class AlertSimulateResponseSerializer(serializers.Serializer):
    data = serializers.ListField(child=serializers.FloatField(), help_text="Data values for each point.")  # type: ignore[assignment]
    dates = serializers.ListField(child=serializers.CharField(), help_text="Date labels for each point.")
    scores = serializers.ListField(
        child=serializers.FloatField(allow_null=True),
        help_text="Anomaly score for each point (null if insufficient data).",
    )
    triggered_indices = serializers.ListField(
        child=serializers.IntegerField(), help_text="Indices of points flagged as anomalies."
    )
    triggered_dates = serializers.ListField(
        child=serializers.CharField(), help_text="Dates of points flagged as anomalies."
    )
    interval = serializers.CharField(
        allow_null=True, help_text="Interval of the trends query (hour, day, week, month)."
    )
    total_points = serializers.IntegerField(help_text="Total number of data points analyzed.")
    anomaly_count = serializers.IntegerField(help_text="Number of anomalies detected.")
    sub_detector_scores = serializers.ListField(
        child=serializers.DictField(),
        required=False,
        help_text="Per-sub-detector scores for ensemble detectors. Each entry has 'type' and 'scores' fields.",
    )
    breakdown_results = BreakdownSimulationResultSerializer(
        many=True,
        required=False,
        help_text=f"Per-breakdown-value simulation results. Present only when the insight has breakdowns (up to {MAX_DETECTOR_BREAKDOWN_VALUES} values).",
    )


class AlertTestDeliveryResponseSerializer(serializers.Serializer):
    destination_count = serializers.IntegerField(help_text="Number of active destinations queued for test delivery.")
    email_recipient_count = serializers.IntegerField(help_text="Number of subscribed users sent a test email.")
    failed_delivery_channels = serializers.ListField(
        child=serializers.ChoiceField(choices=("email", "destination")),
        help_text="Configured delivery channels that failed to schedule or send.",
    )


@extend_schema_view(
    list=extend_schema(
        parameters=[
            OpenApiParameter(
                "search",
                OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                description=(
                    "Optional. Fuzzy match against alert `name` using Postgres trigram word similarity "
                    "(handles typos, transpositions, and prefix-as-you-type). Results are ordered by relevance, "
                    "then creation time. Capped at 200 characters; longer queries return a 400 error."
                ),
            ),
            OpenApiParameter(
                "created_by",
                OpenApiTypes.UUID,
                location=OpenApiParameter.QUERY,
                description="Optional. Restrict results to alerts created by the user with this UUID.",
            ),
            OpenApiParameter(
                "insight_id",
                OpenApiTypes.INT,
                location=OpenApiParameter.QUERY,
                description="Optional. Restrict results to alerts on this insight ID.",
            ),
        ],
    ),
)
class AlertViewSet(TeamAndOrgViewSetMixin, viewsets.ModelViewSet):
    scope_object = "alert"
    queryset = (
        AlertConfiguration.objects.select_related("team", "insight", "threshold", "created_by")
        .prefetch_related("subscribed_users")
        .order_by("-created_at")
    )
    serializer_class = AlertSerializer

    def safely_get_queryset(self, queryset) -> QuerySet:
        filters = self.request.query_params
        if "insight" in filters:
            queryset = queryset.filter(insight_id=filters["insight"])

        insight_id = filters.get("insight_id")
        if insight_id is not None:
            queryset = queryset.filter(insight_id=insight_id)

        created_by = filters.get("created_by")
        if created_by:
            try:
                uuid.UUID(created_by)
            except ValueError:
                raise ValidationError({"created_by": ["Not a valid UUID."]}) from None
            queryset = queryset.filter(created_by__uuid=created_by)

        search = filters.get("search")
        if search:
            if len(search) > MAX_SEARCH_LENGTH:
                raise serializers.ValidationError(
                    {"search": f"Search query must be {MAX_SEARCH_LENGTH} characters or fewer."}
                )
            queryset = self._apply_search(queryset, search)

        # Gate on viewer access to the linked insight. Alert read/write access does not imply access
        # to the insight's results (carried in AlertCheck.triggered_metadata / calculated_value), and
        # "alert" isn't an access-control resource — so without this an alert leaks a restricted
        # insight's data via check history, or via a PATCH that omits `insight` (skipping the
        # create-time check). Filtering here covers list, retrieve, update, delete, and the embedded
        # checks, since get_object() resolves detail routes from this queryset.
        viewable_insights = self.user_access_control.filter_queryset_by_access_level(
            Insight.objects.filter(team_id=self.team_id)
        )
        queryset = queryset.filter(insight_id__in=viewable_insights.values("id"))

        # Read-side twin of the write-side metrics scope gate: a programmatic token without the
        # metrics data scope must not read metric alert results (last_value, check history, breach
        # labels) that a session user scheduled. Filtering here covers list, retrieve, update,
        # delete, and the embedded checks, since get_object() resolves from this queryset.
        # filter-then-exclude-by-id rather than exclude(Q) — NOT over a JSON path is three-valued,
        # so a plain exclude also drops rows whose query lacks the key entirely.
        if _token_lacks_metrics_scope(self.request):
            queryset = queryset.exclude(id__in=queryset.filter(_METRICS_INSIGHT_QUERY_FILTER).values("id"))

        latest_check = AlertCheck.objects.filter(alert_configuration=OuterRef("pk")).order_by("-created_at")
        queryset = queryset.annotate(last_value=Subquery(latest_check.values("calculated_value")[:1]))

        return queryset

    def filter_queryset(self, queryset: QuerySet) -> QuerySet:
        return drop_similar_when_exact_exists(super().filter_queryset(queryset))

    @staticmethod
    def _apply_search(queryset: QuerySet, search: str) -> QuerySet:
        return apply_trigram_search(
            queryset,
            search,
            span_prefix="alert.search",
            fields=(NAME_FIELD,),
            tiebreakers=("-created_at",),
        )

    CHECKS_DEFAULT_LIMIT = 5
    CHECKS_MAX_LIMIT = 500

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="checks_date_from",
                type=str,
                required=False,
                description="Relative date string for the start of the check history window (e.g. '-24h', '-7d', '-14d'). Returns checks created after this time. Max retention is 14 days.",
            ),
            OpenApiParameter(
                name="checks_date_to",
                type=str,
                required=False,
                description="Relative date string for the end of the check history window (e.g. '-1h', '-1d'). Defaults to now if not specified.",
            ),
            OpenApiParameter(
                name="checks_limit",
                type=int,
                required=False,
                description="Maximum number of check results to return (default 5, max 500). Applied after date filtering.",
            ),
            OpenApiParameter(
                name="checks_offset",
                type=int,
                required=False,
                description="Number of newest checks to skip (0-based). Use with checks_limit for pagination. Default 0.",
            ),
        ],
    )
    def retrieve(self, request, *args, **kwargs):
        instance = self.get_object()

        checks_qs = instance.alertcheck_set.select_related("investigation_notebook").order_by("-created_at")

        checks_date_from = request.query_params.get("checks_date_from")
        if checks_date_from:
            parsed_date = relative_date_parse(checks_date_from, ZoneInfo("UTC"))
            checks_qs = checks_qs.filter(created_at__gte=parsed_date)

        checks_date_to = request.query_params.get("checks_date_to")
        if checks_date_to:
            parsed_date = relative_date_parse(checks_date_to, ZoneInfo("UTC"))
            checks_qs = checks_qs.filter(created_at__lte=parsed_date)

        has_date_filter = checks_date_from or checks_date_to
        raw_limit = request.query_params.get("checks_limit")
        if raw_limit is not None:
            try:
                limit = max(1, min(int(raw_limit), self.CHECKS_MAX_LIMIT))
            except (ValueError, TypeError):
                limit = self.CHECKS_DEFAULT_LIMIT
        else:
            limit = self.CHECKS_MAX_LIMIT if has_date_filter else self.CHECKS_DEFAULT_LIMIT

        raw_offset = request.query_params.get("checks_offset")
        if raw_offset is not None:
            try:
                offset = max(0, int(raw_offset))
            except (ValueError, TypeError):
                offset = 0
        else:
            offset = 0

        checks_total = checks_qs.count()
        instance.checks_total = checks_total
        offset = min(offset, checks_total)
        instance.checks = checks_qs[offset : offset + limit]
        serializer = self.get_serializer(instance)
        return Response(serializer.data)

    def list(self, request, *args, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())

        # Paginate first, then prefetch checks only for the page
        page = self.paginate_queryset(queryset)
        alerts = list(page) if page is not None else list(queryset)

        # Prefetch firing checks for anomaly point display on chart.
        if alerts:
            alert_ids = [a.id for a in alerts]
            firing_checks = (
                AlertCheck.objects.filter(
                    alert_configuration_id__in=alert_ids,
                    triggered_points__isnull=False,
                )
                .exclude(triggered_points=[])
                .select_related("investigation_notebook")
                .order_by("-created_at")
            )
            checks_by_alert: dict[str, list] = {str(a.id): [] for a in alerts}
            for check in firing_checks:
                checks_by_alert.setdefault(str(check.alert_configuration_id), []).append(check)
            for alert in alerts:
                alert.checks = checks_by_alert.get(str(alert.id), [])

        if page is not None:
            serializer = self.get_serializer(alerts, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(alerts, many=True)
        return Response(serializer.data)

    @extend_schema(
        request=None,
        responses={202: AlertTestDeliveryResponseSerializer},
        description="Send a synthetic test notification to subscribed users and every active destination on this alert.",
    )
    @action(
        detail=True,
        methods=["POST"],
        url_path="test-delivery",
        required_scopes=["alert:write"],
        throttle_classes=[AlertTestDeliveryThrottle],
    )
    def test_delivery(self, request, *args, **kwargs):
        alert = self.get_object()
        destination_count = count_active_alert_destinations(
            team_id=alert.team_id,
            alert_id=str(alert.id),
            allowed_event_ids=(INSIGHT_ALERT_FIRING_EVENT,),
        )
        email_targets = alert.get_subscribed_users_emails()
        if destination_count == 0 and not email_targets:
            return Response(
                {"detail": "Add an email recipient or active destination before sending a test."},
                status=status.HTTP_409_CONFLICT,
            )

        failed_delivery_channels: list[str] = []
        successful_email_count = 0
        successful_destination_count = 0
        if email_targets:
            try:
                send_test_alert_email(alert, recipients=email_targets, idempotency_key=str(uuid.uuid4()))
                successful_email_count = len(email_targets)
            except Exception as error:
                capture_exception(
                    error,
                    additional_properties={"alert_id": str(alert.id), "feature": "alerts", "channel": "email"},
                )
                failed_delivery_channels.append("email")
        if destination_count and trigger_alert_insights_functions(
            alert,
            {
                "breaches": "Test alert from Insights. No action is needed.",
                "is_test": True,
                "alert_name": f"[TEST] {alert.name}",
            },
        ):
            successful_destination_count = destination_count
        elif destination_count:
            failed_delivery_channels.append("destination")

        if successful_email_count == 0 and successful_destination_count == 0:
            return Response(
                {"detail": "Unable to start the test delivery. Check the configured channels and try again."},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )
        hanzo_insights.capture(
            distinct_id=str(request.user.distinct_id),
            event="insight alert test delivery scheduled",
            properties={
                **get_request_analytics_properties(request),
                "alert_id": str(alert.id),
                "destination_count": successful_destination_count,
                "email_recipient_count": successful_email_count,
                "failed_delivery_channels": failed_delivery_channels,
                "team_id": alert.team_id,
            },
        )
        return Response(
            {
                "destination_count": successful_destination_count,
                "email_recipient_count": successful_email_count,
                "failed_delivery_channels": failed_delivery_channels,
            },
            status=status.HTTP_202_ACCEPTED,
        )

    @extend_schema(
        request=AlertSimulateSerializer,
        responses={200: AlertSimulateResponseSerializer},
        description="Simulate a detector on an insight's historical data. Read-only — no AlertCheck records are created.",
    )
    # Returns an insight's computed result series, so it requires insight read in addition to
    # alert read — an alert-scoped token must not read insight/query data it isn't scoped for.
    # (Object-level insight viewer access is enforced separately in AlertSimulateSerializer.)
    @action(detail=False, methods=["POST"], url_path="simulate", required_scopes=["alert:read", "insight:read"])
    def simulate(self, request, *args, **kwargs):
        serializer = AlertSimulateSerializer(data=request.data, context=self.get_serializer_context())
        serializer.is_valid(raise_exception=True)

        insight = serializer.validated_data["insight"]
        detector_config = serializer.validated_data["detector_config"]
        series_index = serializer.validated_data["series_index"]
        date_from = serializer.validated_data.get("date_from")
        config = serializer.validated_data.get("config")

        try:
            result = simulate_detector_on_insight(
                insight=insight,
                team=self.team,
                detector_config=detector_config,
                series_index=series_index,
                date_from=date_from,
                user=cast(User, request.user),
                config=config,
            )
        except (ValueError, IndexError, AlertExtractionError) as e:
            raise ValidationError(str(e))
        except RuntimeError:
            raise ValidationError("Simulation failed: unable to compute results for this insight.")

        response_serializer = AlertSimulateResponseSerializer(result)
        return Response(response_serializer.data)


class ThresholdWithAlertSerializer(ThresholdSerializer):
    alerts = AlertSerializer(many=True, read_only=True, source="alertconfiguration_set")

    class Meta(ThresholdSerializer.Meta):
        fields = [*ThresholdSerializer.Meta.fields, "alerts"]
        read_only_fields = [*ThresholdSerializer.Meta.read_only_fields, "alerts"]


class ThresholdViewSet(TeamAndOrgViewSetMixin, viewsets.ReadOnlyModelViewSet):
    scope_object = "alert"
    queryset = Threshold.objects.all()
    serializer_class = ThresholdWithAlertSerializer

    def safely_get_queryset(self, queryset) -> QuerySet:
        # Thresholds embed full alerts (including check values), so they get the same read-side
        # metrics scope gate as the alerts endpoints. Same null-safe filter-then-exclude shape.
        if _token_lacks_metrics_scope(self.request):
            queryset = queryset.exclude(id__in=queryset.filter(_METRICS_INSIGHT_QUERY_FILTER).values("id"))
        return queryset