insights/products/surveys/backend/api/survey.py

import re
import builtins
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta
from typing import Any, Optional, cast
from urllib.parse import urlparse
from uuid import UUID

from django.conf import settings
from django.core.cache import cache
from django.db import transaction
from django.db.models import Min, QuerySet
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render
from django.utils.text import slugify
from django.views.decorators.csrf import csrf_exempt

import nh3
import structlog
import django_filters
import hanzo_insights
from django_filters.rest_framework import DjangoFilterBackend, FilterSet
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import (
    OpenApiParameter,
    PolymorphicProxySerializer,
    extend_schema,
    extend_schema_field,
    extend_schema_view,
    inline_serializer,
)
from hanzo_insights import capture_exception
from nanoid import generate
from opentelemetry import trace
from rest_framework import exceptions, request, serializers, status, viewsets
from rest_framework.request import Request
from rest_framework.response import Response

from insights.schema import ProductKey

from insights.api.documentation import FeatureFlagFiltersSchemaSerializer
from insights.api.routing import TeamAndOrgViewSetMixin
from insights.api.shared import SearchMatchTypeSerializerMixin, UserBasicSerializer
from insights.api.utils import action
from insights.cloud_utils import is_cloud
from insights.constants import SURVEY_TARGETING_FLAG_PREFIX, AvailableFeature
from insights.datastore.client import sync_execute
from insights.datastore.query_tagging import Feature, tag_queries
from insights.event_usage import report_user_action
from insights.helpers.impersonation import is_impersonated
from insights.helpers.trigram_search import (
    DESCRIPTION_FIELD,
    MAX_SEARCH_LENGTH,
    NAME_FIELD,
    apply_trigram_search,
    drop_similar_when_exact_exists,
)
from insights.models.activity_logging.activity_log import Change, Detail, changes_between, load_activity, log_activity
from insights.models.activity_logging.activity_page import activity_page_response
from insights.models.team.team import Team
from insights.models.user import User
from insights.models.utils import UUIDT
from insights.rbac.access_control_api_mixin import AccessControlViewSetMixin
from insights.rbac.user_access_control import UserAccessControlSerializerMixin
from insights.utils_cors import cors_response

from products.actions.backend.api.action import ActionSerializer, ActionStepJSONSerializer
from products.actions.backend.models.action import Action
from products.feature_flags.backend.api.feature_flag import (
    BEHAVIOURAL_COHORT_FOUND_ERROR_CODE,
    FeatureFlagSerializer,
    MinimalFeatureFlagSerializer,
    assert_feature_flag_write_scope,
)
from products.feature_flags.backend.models.feature_flag import FeatureFlag
from products.product_analytics.backend.models.insight import Insight
from products.surveys.backend.models import MAX_ITERATION_COUNT, Survey, SurveyResponseArchive, ensure_question_ids
from products.surveys.backend.responses import (
    SurveyRates,
    SurveyStats,
    archived_responses_filter,
    build_choice_translation_map,
    calculate_rates,
    fetch_per_question_stats,
    fetch_response_rows,
    get_survey_stats,
    partial_responses_filter,
    process_survey_results,
    validate_and_parse_dates,
)
from products.surveys.backend.summarization import fetch_responses, format_as_markdown, summarize_responses
from products.surveys.backend.translation import generate_survey_translation
from products.surveys.backend.util import (
    SurveyEventName,
    SurveyEventProperties,
    get_archived_response_uuids,
    get_survey_property_string_expr,
)

# Constants for better maintainability
logger = structlog.get_logger(__name__)
tracer = trace.get_tracer(__name__)
CACHE_TIMEOUT_SECONDS = 300
DISPLAY_LANGUAGE_QUERY_PARAM = "display_language"
DISPLAY_LANGUAGE_RE = re.compile(r"^[A-Za-z]{2,3}(-[A-Za-z0-9]{2,8}){0,3}$")

ALLOWED_LINK_URL_SCHEMES = ["https", "mailto"]
EMAIL_REGEX = r"^mailto:[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"

# Translation language codes must be BCP-47-ish (lang[-subtag...]). This is the same shape
# the JS SDK matches against navigator.language. Aliases like "english" or sentinel values
# like "default" are rejected because they never match a real browser locale.
BCP47_LANGUAGE_CODE_RE = re.compile(r"^[a-z]{2,3}(-[a-z0-9]{2,8}){0,3}$")
DEFAULT_BASE_LANGUAGE = "en"
# Sentinel keys we explicitly reject — these used to appear in customer data and never
# resolved to anything in the SDK. Block them at the API so they don't keep accumulating.
REJECTED_TRANSLATION_KEYS = frozenset({"default", "original", "base"})


def _normalize_language_code(raw: str) -> str:
    """Lowercase + underscore-to-hyphen. Matches what the SDK does before lookup."""
    return raw.strip().lower().replace("_", "-")


# Keep this in sync with SurveyAPISerializer's public runtime contract.
# Root survey description is intentionally excluded because customers have used it for internal notes.
SURVEY_API_TRANSLATION_FIELDS = frozenset(
    [
        "name",
        "thankYouMessageHeader",
        "thankYouMessageDescription",
        "thankYouMessageCloseButtonText",
    ]
)
FIELDS_NOT_APPLICABLE_TO_EXTERNAL_SURVEYS = [
    "linked_flag_id",
    "targeting_flag_filters",
]
SURVEY_TRANSLATION_DRAFT_FIELDS = ("name", "description", "type", "appearance", "questions", "translations")
SURVEY_TRANSLATION_DRAFT_APPEARANCE_FIELDS = (
    "thankYouMessageHeader",
    "thankYouMessageDescription",
    "thankYouMessageCloseButtonText",
)
SURVEY_TRANSLATION_DRAFT_QUESTION_FIELDS = (
    "id",
    "type",
    "question",
    "description",
    "buttonText",
    "choices",
    "lowerBoundLabel",
    "upperBoundLabel",
    "link",
    "translations",
)


class GenerateSurveyTranslationsRequestSerializer(serializers.Serializer):
    target_language = serializers.CharField(help_text="Language code to generate translations for, for example pt-BR.")
    source_language = serializers.CharField(
        required=False,
        allow_blank=True,
        help_text=(
            "Optional override for the source language code. Defaults to the survey's `base_language` "
            "(or 'en' if unset)."
        ),
    )
    overwrite = serializers.BooleanField(
        required=False, default=False, help_text="Whether to overwrite existing translations for this language."
    )
    survey = serializers.DictField(
        child=serializers.JSONField(allow_null=True, help_text="Draft survey field value."),
        required=False,
        help_text="Optional translation-only draft survey payload to translate instead of the last saved survey.",
    )

    def validate_survey(self, survey: dict[str, Any]) -> dict[str, Any]:
        draft = {field: survey[field] for field in SURVEY_TRANSLATION_DRAFT_FIELDS if field in survey}

        appearance = draft.get("appearance")
        if isinstance(appearance, dict):
            draft["appearance"] = {
                field: appearance[field] for field in SURVEY_TRANSLATION_DRAFT_APPEARANCE_FIELDS if field in appearance
            }

        questions = draft.get("questions")
        if isinstance(questions, list):
            draft["questions"] = [
                {field: question[field] for field in SURVEY_TRANSLATION_DRAFT_QUESTION_FIELDS if field in question}
                for question in questions
                if isinstance(question, dict)
            ]

        return draft


class GeneratedSurveyRootTranslationSerializer(serializers.Serializer):
    name = serializers.CharField(required=False, allow_blank=True, help_text="Translated survey name.")
    thankYouMessageHeader = serializers.CharField(
        required=False, allow_blank=True, help_text="Translated thank-you header."
    )
    thankYouMessageDescription = serializers.CharField(
        required=False, allow_blank=True, help_text="Translated thank-you description."
    )
    thankYouMessageCloseButtonText = serializers.CharField(
        required=False, allow_blank=True, help_text="Translated thank-you close button text."
    )


class GeneratedSurveyQuestionTranslationSerializer(serializers.Serializer):
    question = serializers.CharField(required=False, allow_blank=True, help_text="Translated question text.")
    description = serializers.CharField(required=False, allow_blank=True, help_text="Translated question description.")
    buttonText = serializers.CharField(required=False, allow_blank=True, help_text="Translated submit button text.")
    choices = serializers.ListField(
        child=serializers.CharField(allow_blank=True),
        required=False,
        help_text="Translated choices in the same order as the source choices.",
    )
    lowerBoundLabel = serializers.CharField(
        required=False, allow_blank=True, help_text="Translated lower rating bound label."
    )
    upperBoundLabel = serializers.CharField(
        required=False, allow_blank=True, help_text="Translated upper rating bound label."
    )
    link = serializers.CharField(required=False, allow_blank=True, help_text="Translated link text or localized URL.")


class GeneratedSurveyQuestionTranslationPatchSerializer(serializers.Serializer):
    id = serializers.CharField(help_text="Survey question id this patch applies to.")
    translations = serializers.DictField(
        child=GeneratedSurveyQuestionTranslationSerializer(),
        help_text="Question translation patch keyed by target language.",
    )


class GenerateSurveyTranslationsResponseSerializer(serializers.Serializer):
    translations = serializers.DictField(
        child=GeneratedSurveyRootTranslationSerializer(),
        help_text="Survey-level translation patch keyed by language.",
    )
    questions = serializers.ListField(
        child=GeneratedSurveyQuestionTranslationPatchSerializer(),
        help_text="Question-level translation patches keyed by question id and language.",
    )
    generated_field_paths = serializers.ListField(
        child=serializers.CharField(),
        help_text="Editor field paths generated by AI and safe to highlight as draft content.",
    )
    trace_id = serializers.CharField(help_text="LLM trace id for debugging and feedback.")


def get_hosted_survey_display_language(request: HttpRequest) -> str | None:
    display_language = request.GET.get(DISPLAY_LANGUAGE_QUERY_PARAM)
    if not display_language:
        return None

    display_language = display_language.strip()
    if not display_language or len(display_language) > 35 or not DISPLAY_LANGUAGE_RE.fullmatch(display_language):
        return None

    return display_language


# Does not include actions or events, as those are objects and thus are evaluated differently
CONDITION_FIELDS_NOT_APPLICABLE_TO_EXTERNAL_SURVEYS = [
    "url",
    "urlMatchType",
    "selector",
    "seenSurveyWaitPeriodInDays",
    "linkedFlagVariant",
    "deviceTypes",
    "deviceTypesMatchType",
]

if "replica" in settings.DATABASES:
    READ_DB_FOR_SURVEYS = "replica"
else:
    READ_DB_FOR_SURVEYS = "default"


# EventStats / SurveyRates / SurveyStats and the stats-computation helpers now live in
# products.surveys.backend.responses.stats so the survey widget shares one query path. They are
# re-imported above and re-exported here for existing importers.


def get_survey_conditions_with_actions(
    survey: "Survey", action_serializer_class: type[serializers.Serializer] | None = None
) -> dict | None:
    if action_serializer_class is None:
        action_serializer_class = ActionSerializer
    conditions = survey.conditions
    actions = survey.actions.all()
    if len(actions) > 0:
        if conditions is None:
            conditions = {}
        else:
            conditions = dict(conditions)
        conditions["actions"] = {"values": action_serializer_class(actions, many=True).data}
    elif isinstance(conditions, dict) and isinstance(conditions.get("actions"), dict):
        # The actions M2M didn't resolve (e.g. the referenced actions were deleted) but a stale
        # actions blob still lives in `conditions`. That stored blob can carry fields this caller's
        # serializer would never expose — notably `created_by` on the public /surveys and /decide
        # payload — so project each value down to the serializer's own fields. Never surface more
        # than the serializer itself would.
        meta = getattr(action_serializer_class, "Meta", None)
        allowed_fields = getattr(meta, "fields", None)
        if isinstance(allowed_fields, list | tuple):
            allowed = set(allowed_fields)
            values = conditions["actions"].get("values") or []
            conditions = dict(conditions)
            conditions["actions"] = {
                "values": [
                    {k: v for k, v in value.items() if k in allowed} for value in values if isinstance(value, dict)
                ]
            }
    return conditions


@extend_schema_field(
    {
        "oneOf": [
            {"type": "integer", "minimum": 0},
            {"type": "string", "enum": ["end"]},
        ]
    }
)
class SurveyBranchTargetField(serializers.Field):
    def to_internal_value(self, data):
        return data

    def to_representation(self, value):
        return value


class SurveyNextQuestionBranchingSerializer(serializers.Serializer):
    type = serializers.ChoiceField(
        choices=["next_question"],
        required=True,
        help_text="Continue to the next question in sequence.",
    )


class SurveyEndBranchingSerializer(serializers.Serializer):
    type = serializers.ChoiceField(choices=["end"], required=True, help_text="End the survey.")


class SurveySpecificQuestionBranchingSerializer(serializers.Serializer):
    type = serializers.ChoiceField(
        choices=["specific_question"],
        required=True,
        help_text="Jump to a specific question index.",
    )
    index = serializers.IntegerField(required=True, min_value=0, help_text="0-based index of the next question.")


class SurveyResponseBasedBranchingSerializer(serializers.Serializer):
    type = serializers.ChoiceField(
        choices=["response_based"],
        required=True,
        help_text="Branch based on the selected or entered response.",
    )
    responseValues = serializers.DictField(
        required=True,
        child=SurveyBranchTargetField(),
        help_text="Response-based branching map. Values can be a question index or 'end'.",
    )


_SurveyBranchingUnion = PolymorphicProxySerializer(
    component_name="SurveyBranchingSchema",
    serializers=[
        SurveyNextQuestionBranchingSerializer,
        SurveyEndBranchingSerializer,
        SurveySpecificQuestionBranchingSerializer,
        SurveyResponseBasedBranchingSerializer,
    ],
    resource_type_field_name=None,
)


@extend_schema_field(_SurveyBranchingUnion)
class SurveyBranchingSchemaField(serializers.Field):
    def to_internal_value(self, data):
        return data

    def to_representation(self, value):
        return value


class SurveyQuestionValidationRuleSerializer(serializers.Serializer):
    type = serializers.ChoiceField(
        choices=["min_length", "max_length", "email"],
        required=True,
        help_text="Validation rule type for open-text questions.",
    )
    value = serializers.IntegerField(
        required=False,
        min_value=1,
        allow_null=True,
        help_text="Rule value for min_length or max_length validations.",
    )


class SurveyBaseQuestionSchemaSerializer(serializers.Serializer):
    id = serializers.CharField(
        required=False,
        help_text=(
            "Stable question identifier (UUID). When editing an existing question, send back its current id so "
            "its responses (keyed by $survey_response_<id>) stay attached; omit it for new questions and the "
            "server generates one."
        ),
    )
    type = serializers.ChoiceField(
        choices=["open", "link", "rating", "single_choice", "multiple_choice"],
        required=True,
        help_text="Question type.",
    )
    question = serializers.CharField(required=True, help_text="Question text shown to respondents.")
    description = serializers.CharField(required=False, allow_blank=True, help_text="Optional helper text.")
    descriptionContentType = serializers.ChoiceField(
        choices=["text", "html"],
        required=False,
        help_text="Format for the description field.",
    )
    optional = serializers.BooleanField(required=False, help_text="Whether respondents may skip this question.")
    buttonText = serializers.CharField(required=False, allow_blank=True, help_text="Custom button label.")


class SurveyOpenQuestionSchemaSerializer(SurveyBaseQuestionSchemaSerializer):
    type = serializers.ChoiceField(choices=["open"], required=True)


class SurveyLinkQuestionSchemaSerializer(SurveyBaseQuestionSchemaSerializer):
    type = serializers.ChoiceField(choices=["link"], required=True)
    link = serializers.CharField(required=True, help_text="HTTPS or mailto URL for link questions.")


class SurveyRatingQuestionSchemaSerializer(SurveyBaseQuestionSchemaSerializer):
    type = serializers.ChoiceField(choices=["rating"], required=True)
    display = serializers.ChoiceField(
        choices=["number", "emoji"],
        required=False,
        help_text="Display format: 'number' shows numeric scale, 'emoji' shows emoji scale.",
    )
    scale = serializers.IntegerField(
        required=False,
        min_value=1,
        help_text="Rating scale can be one of 3, 5, or 7",
    )
    lowerBoundLabel = serializers.CharField(
        required=False,
        allow_blank=True,
        help_text="Label for the lowest rating (e.g., 'Very Poor')",
    )
    upperBoundLabel = serializers.CharField(
        required=False,
        allow_blank=True,
        help_text="Label for the highest rating (e.g., 'Excellent')",
    )
    branching = SurveyBranchingSchemaField(required=False, allow_null=True)


class SurveySingleChoiceQuestionSchemaSerializer(SurveyBaseQuestionSchemaSerializer):
    type = serializers.ChoiceField(choices=["single_choice"], required=True)
    choices = serializers.ListField(
        child=serializers.CharField(),
        required=True,
        min_length=2,
        max_length=20,
        help_text="Array of choice options. Choice indices (0, 1, 2, ...) are used for branching logic.",
    )
    shuffleOptions = serializers.BooleanField(
        required=False,
        help_text="Whether to randomize the order of choices for each respondent.",
    )
    hasOpenChoice = serializers.BooleanField(
        required=False,
        help_text="Whether the final option should be an open-text choice (for example, 'Other').",
    )
    branching = SurveyBranchingSchemaField(required=False, allow_null=True)


class SurveyMultipleChoiceQuestionSchemaSerializer(SurveyBaseQuestionSchemaSerializer):
    type = serializers.ChoiceField(choices=["multiple_choice"], required=True)
    choices = serializers.ListField(
        child=serializers.CharField(),
        required=True,
        min_length=2,
        max_length=20,
        help_text="Array of choice options. Multiple selections allowed. No branching logic supported.",
    )
    shuffleOptions = serializers.BooleanField(
        required=False,
        help_text="Whether to randomize the order of choices for each respondent.",
    )
    hasOpenChoice = serializers.BooleanField(
        required=False,
        help_text="Whether the final option should be an open-text choice (for example, 'Other').",
    )


_SurveyQuestionUnion = PolymorphicProxySerializer(
    component_name="SurveyQuestionInputSchema",
    serializers=[
        SurveyOpenQuestionSchemaSerializer,
        SurveyLinkQuestionSchemaSerializer,
        SurveyRatingQuestionSchemaSerializer,
        SurveySingleChoiceQuestionSchemaSerializer,
        SurveyMultipleChoiceQuestionSchemaSerializer,
    ],
    resource_type_field_name=None,
)


@extend_schema_field(serializers.ListSerializer(child=_SurveyQuestionUnion))
class SurveyQuestionsSchemaField(serializers.ListField):
    pass


SURVEY_MATCH_TYPE_CHOICES = ["regex", "not_regex", "exact", "is_not", "icontains", "not_icontains"]


class SurveyAppearanceSchemaSerializer(serializers.Serializer):
    backgroundColor = serializers.CharField(required=False)
    submitButtonColor = serializers.CharField(required=False)
    textColor = serializers.CharField(required=False)
    submitButtonText = serializers.CharField(required=False)
    submitButtonTextColor = serializers.CharField(required=False)
    descriptionTextColor = serializers.CharField(required=False)
    ratingButtonColor = serializers.CharField(required=False)
    ratingButtonActiveColor = serializers.CharField(required=False)
    ratingButtonHoverColor = serializers.CharField(required=False)
    whiteLabel = serializers.BooleanField(required=False)
    autoDisappear = serializers.BooleanField(required=False)
    displayThankYouMessage = serializers.BooleanField(required=False)
    thankYouMessageHeader = serializers.CharField(required=False)
    thankYouMessageDescription = serializers.CharField(required=False)
    thankYouMessageDescriptionContentType = serializers.ChoiceField(
        choices=["html", "text"],
        required=False,
    )
    thankYouMessageCloseButtonText = serializers.CharField(required=False)
    borderColor = serializers.CharField(required=False)
    placeholder = serializers.CharField(required=False)
    shuffleQuestions = serializers.BooleanField(required=False)
    surveyPopupDelaySeconds = serializers.IntegerField(required=False)
    allowGoBack = serializers.BooleanField(
        required=False,
        help_text="Whether to show a 'Back' button on web surveys after the first question, letting respondents return to a previously visited question. Defaults to false.",
    )
    backButtonText = serializers.CharField(
        required=False,
        help_text="Optional override for the back button label. Defaults to 'Back'.",
    )
    widgetType = serializers.ChoiceField(choices=["button", "tab", "selector"], required=False)
    widgetSelector = serializers.CharField(required=False)
    widgetLabel = serializers.CharField(required=False)
    widgetColor = serializers.CharField(required=False)
    fontFamily = serializers.CharField(required=False)
    maxWidth = serializers.CharField(required=False)
    zIndex = serializers.CharField(required=False)
    disabledButtonOpacity = serializers.CharField(required=False)
    boxPadding = serializers.CharField(required=False)


SURVEY_DEVICE_TYPE_CHOICES = ["Desktop", "Mobile", "Tablet"]
SURVEY_MATCH_TYPE_HELP_TEXT = (
    "URL/device matching types: 'regex' (matches regex pattern), 'not_regex' (does not match regex pattern), "
    "'exact' (exact string match), 'is_not' (not exact match), 'icontains' (case-insensitive contains), "
    "'not_icontains' (case-insensitive does not contain)."
)


class SurveyConditionEventValueSchemaSerializer(serializers.Serializer):
    name = serializers.CharField(help_text="Event name that triggers the survey.")


class SurveyEventsConditionSchemaSerializer(serializers.Serializer):
    repeatedActivation = serializers.BooleanField(
        required=False,
        help_text="Whether to show the survey every time one of the events is triggered (true), or just once (false).",
    )
    values = SurveyConditionEventValueSchemaSerializer(
        many=True,
        required=False,
        help_text="Array of event names that trigger the survey.",
    )


class SurveyConditionsSchemaSerializer(serializers.Serializer):
    url = serializers.CharField(required=False, allow_blank=True)
    selector = serializers.CharField(required=False, allow_blank=True)
    seenSurveyWaitPeriodInDays = serializers.IntegerField(
        required=False,
        min_value=0,
        help_text="Don't show this survey to users who saw any survey in the last x days.",
    )
    urlMatchType = serializers.ChoiceField(
        required=False,
        choices=SURVEY_MATCH_TYPE_CHOICES,
        help_text=SURVEY_MATCH_TYPE_HELP_TEXT,
    )
    events = SurveyEventsConditionSchemaSerializer(required=False)
    deviceTypes = serializers.ListField(
        required=False,
        child=serializers.ChoiceField(choices=SURVEY_DEVICE_TYPE_CHOICES),
        help_text="Device types that should match for this survey to be shown.",
    )
    deviceTypesMatchType = serializers.ChoiceField(
        required=False,
        choices=SURVEY_MATCH_TYPE_CHOICES,
        help_text=SURVEY_MATCH_TYPE_HELP_TEXT,
    )
    linkedFlagVariant = serializers.CharField(
        required=False,
        allow_blank=True,
        help_text="The variant of the feature flag linked to this survey.",
    )


class SurveySummarizeRequestSerializer(serializers.Serializer):
    force_refresh = serializers.BooleanField(
        required=False,
        default=False,
        help_text="When true, bypass cached summaries and regenerate. Defaults to false.",
    )


class SurveyResponseAnswerSerializer(serializers.Serializer):
    question_id = serializers.CharField(help_text="UUID of the survey question this answer belongs to.")
    question_index = serializers.IntegerField(help_text="Zero-based index of the question within the survey.")
    question_text = serializers.CharField(
        allow_blank=True, help_text="Untranslated question text as configured by the survey author."
    )
    question_type = serializers.CharField(
        help_text=(
            "Question type: open, rating, single_choice, multiple_choice, or link. "
            "Determines the shape of the answer field."
        ),
    )
    answer = serializers.JSONField(
        help_text=(
            "Resolved answer. String for open/rating/single_choice/link questions, "
            "list of strings for multiple_choice questions. Already decoded from the "
            "raw $survey_response_<id> property so callers don't need to parse it."
        )
    )


class SurveyResponseExtraSerializer(serializers.Serializer):
    device_type = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$device_type at the time the response was sent."
    )
    browser = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$browser at the time the response was sent."
    )
    os = serializers.CharField(
        allow_null=True,
        allow_blank=True,
        required=False,
        help_text="$os (operating system) at the time the response was sent.",
    )
    geoip_country_code = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$geoip_country_code at submission time."
    )
    geoip_country_name = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$geoip_country_name at submission time."
    )
    geoip_city_name = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$geoip_city_name at submission time."
    )
    current_url = serializers.CharField(
        allow_null=True, allow_blank=True, required=False, help_text="$current_url where the survey was submitted."
    )
    iteration = serializers.CharField(
        allow_null=True,
        allow_blank=True,
        required=False,
        help_text="Survey iteration number when the response was sent. Only set for recurring surveys.",
    )


class SurveyResponseRowSerializer(serializers.Serializer):
    uuid = serializers.CharField(
        help_text="UUID of the underlying `survey sent` event. Use as the response identifier for archive operations."
    )
    distinct_id = serializers.CharField(
        help_text="distinct_id of the respondent. Cross-pivot to the persons API or session recordings."
    )
    session_id = serializers.CharField(
        allow_null=True,
        help_text="$session_id of the respondent when available. Use to pull the session recording for this response.",
    )
    submitted_at = serializers.DateTimeField(help_text="Event timestamp when the response was sent (ISO 8601, UTC).")
    answers = SurveyResponseAnswerSerializer(
        many=True,
        help_text=(
            "One entry per survey question that received a non-empty answer. Question text is already resolved — "
            "callers do not need to look up `$survey_response_<id>` keys."
        ),
    )
    extra = SurveyResponseExtraSerializer(
        help_text="Convenience fields extracted from the event properties (device, browser, geoip, iteration)."
    )


class SurveyResponsesQuerySerializer(serializers.Serializer):
    since = serializers.DateTimeField(
        required=False,
        help_text="Only return responses submitted on or after this ISO 8601 timestamp.",
    )
    until = serializers.DateTimeField(
        required=False,
        help_text="Only return responses submitted on or before this ISO 8601 timestamp.",
    )
    question_id = serializers.CharField(
        required=False,
        help_text=(
            "If set, only return rows where this question has a non-empty answer, and only include that question's "
            "answer in each row. Required when using score_lte or score_gte."
        ),
    )
    score_lte = serializers.FloatField(
        required=False,
        help_text=(
            "Filter to rows where the rating answer for `question_id` is <= this value. "
            "Common use: NPS detractors with score_lte=6. Requires question_id."
        ),
    )
    score_gte = serializers.FloatField(
        required=False,
        help_text=(
            "Filter to rows where the rating answer for `question_id` is >= this value. "
            "Common use: NPS promoters with score_gte=9. Requires question_id."
        ),
    )
    exclude_archived = serializers.BooleanField(
        required=False,
        default=False,
        help_text="When true, exclude responses that have been archived via the archive_response endpoint.",
    )
    limit = serializers.IntegerField(
        required=False,
        default=100,
        min_value=1,
        max_value=500,
        help_text="Maximum number of rows to return (1-500). Defaults to 100.",
    )
    offset = serializers.IntegerField(
        required=False,
        default=0,
        min_value=0,
        help_text="Number of rows to skip for pagination. Combine with `limit` and the `has_more` field to paginate.",
    )


class SurveyResponsesListSerializer(serializers.Serializer):
    results = SurveyResponseRowSerializer(many=True, help_text="Survey response rows for the requested page.")
    has_more = serializers.BooleanField(
        help_text="True if more rows exist beyond the current page — fetch the next page with offset + limit."
    )
    limit = serializers.IntegerField(help_text="The limit applied to this query (echoed back for pagination).")
    offset = serializers.IntegerField(help_text="The offset applied to this query (echoed back for pagination).")


class SurveySerializer(SearchMatchTypeSerializerMixin, UserAccessControlSerializerMixin, serializers.ModelSerializer):
    linked_flag_id = serializers.IntegerField(required=False, allow_null=True, source="linked_flag.id")
    linked_flag = MinimalFeatureFlagSerializer(read_only=True)
    linked_insight_id = serializers.IntegerField(required=False, allow_null=True, source="linked_insight.id")
    targeting_flag = MinimalFeatureFlagSerializer(read_only=True)
    internal_targeting_flag = MinimalFeatureFlagSerializer(read_only=True)
    created_by = UserBasicSerializer(read_only=True)
    conditions = serializers.SerializerMethodField(method_name="get_conditions", read_only=True)
    feature_flag_keys = serializers.SerializerMethodField()
    # NB this is enforced in the UI too
    iteration_count = serializers.IntegerField(
        required=False, allow_null=True, max_value=MAX_ITERATION_COUNT, min_value=0
    )
    schedule = serializers.CharField(required=False, allow_null=True)
    enable_partial_responses = serializers.BooleanField(required=False, allow_null=True)
    enable_iframe_embedding = serializers.BooleanField(required=False, allow_null=True)
    base_language = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=20,
        help_text=(
            "BCP-47 language code (e.g. 'en', 'es', 'es-MX') describing the language of the survey's "
            "untranslated text. Defaults to 'en'. Cannot also appear as a key in `translations`."
        ),
    )

    @extend_schema_field(
        serializers.ListField(
            child=serializers.DictField(child=serializers.CharField(allow_null=True)),
        )
    )
    def get_feature_flag_keys(self, survey: Survey) -> list:
        return [
            {"key": "linked_flag_key", "value": survey.linked_flag.key if survey.linked_flag else None},
            {"key": "targeting_flag_key", "value": survey.targeting_flag.key if survey.targeting_flag else None},
            {
                "key": "internal_targeting_flag_key",
                "value": survey.internal_targeting_flag.key if survey.internal_targeting_flag else None,
            },
            {
                "key": "internal_response_sampling_flag_key",
                "value": survey.internal_response_sampling_flag.key if survey.internal_response_sampling_flag else None,
            },
        ]

    class Meta:
        model = Survey
        fields = [
            "id",
            "name",
            "description",
            "type",
            "schedule",
            "linked_flag",
            "linked_flag_id",
            "linked_insight_id",
            "targeting_flag",
            "internal_targeting_flag",
            "questions",
            "conditions",
            "appearance",
            "created_at",
            "created_by",
            "start_date",
            "end_date",
            "archived",
            "responses_limit",
            "feature_flag_keys",
            "iteration_count",
            "iteration_frequency_days",
            "iteration_start_dates",
            "current_iteration",
            "current_iteration_start_date",
            "response_sampling_start_date",
            "response_sampling_interval_type",
            "response_sampling_interval",
            "response_sampling_limit",
            "response_sampling_daily_limits",
            "enable_partial_responses",
            "enable_iframe_embedding",
            "base_language",
            "translations",
            "user_access_level",
            "form_content",
            "search_match_type",
        ]
        read_only_fields = ["id", "created_at", "created_by"]

    @extend_schema_field(serializers.DictField(allow_null=True))
    def get_conditions(self, survey: Survey):
        return get_survey_conditions_with_actions(survey)


class SurveySerializerCreateUpdateOnly(serializers.ModelSerializer):
    linked_flag = MinimalFeatureFlagSerializer(read_only=True)
    linked_flag_id = serializers.IntegerField(required=False, write_only=True, allow_null=True)
    linked_insight_id = serializers.IntegerField(required=False, write_only=True, allow_null=True)
    targeting_flag_id = serializers.IntegerField(required=False, write_only=True)
    targeting_flag_filters = serializers.JSONField(required=False, write_only=True, allow_null=True)

    def _validate_and_sanitize_link(self, link: str) -> str:
        """Validate URL scheme and format, then sanitize HTML. Returns cleaned link."""
        parsed_url = urlparse(link)

        # Check for unsupported schemes
        if parsed_url.scheme not in ALLOWED_LINK_URL_SCHEMES:
            raise serializers.ValidationError(
                f"Link must be a URL with one of these schemes: [{', '.join(ALLOWED_LINK_URL_SCHEMES)}]"
            )

        # Validate mailto links
        if parsed_url.scheme == "mailto":
            if not re.match(EMAIL_REGEX, link):
                raise serializers.ValidationError(
                    "Invalid mailto link. Please enter a valid mailto link (e.g., mailto:[email protected])."
                )
        # Validate HTTPS URLs
        elif parsed_url.scheme == "https":
            if not parsed_url.netloc:
                raise serializers.ValidationError("Invalid HTTPS URL. Please enter a valid HTTPS link.")

        # Sanitize HTML if present
        if nh3.is_html(link):
            return nh3_clean_with_allow_list(link)
        return link

    def _validate_and_sanitize_choices(self, choices: list) -> list:
        """Validate choices are non-empty strings and sanitize HTML. Returns cleaned choices."""
        cleaned_choices = []
        for choice in choices:
            if not isinstance(choice, str):
                raise serializers.ValidationError("Question choices must be strings")
            if not choice.strip():
                raise serializers.ValidationError("Question choices cannot be empty")

            if nh3.is_html(choice):
                cleaned_choices.append(nh3_clean_with_allow_list(choice))
            else:
                cleaned_choices.append(choice)
        return cleaned_choices

    remove_targeting_flag = serializers.BooleanField(required=False, write_only=True, allow_null=True)
    targeting_flag = MinimalFeatureFlagSerializer(read_only=True)
    internal_targeting_flag = MinimalFeatureFlagSerializer(read_only=True)
    created_by = UserBasicSerializer(read_only=True)
    # NB this is enforced in the UI too
    iteration_count = serializers.IntegerField(
        required=False, allow_null=True, max_value=MAX_ITERATION_COUNT, min_value=0
    )
    schedule = serializers.CharField(required=False, allow_null=True)
    enable_partial_responses = serializers.BooleanField(required=False, allow_null=True)
    enable_iframe_embedding = serializers.BooleanField(required=False, allow_null=True)
    base_language = serializers.CharField(
        required=False,
        allow_blank=True,
        max_length=20,
        help_text=(
            "BCP-47 language code (e.g. 'en', 'es', 'es-MX') describing the language of the survey's "
            "untranslated text. Defaults to 'en'. Cannot also appear as a key in `translations`."
        ),
    )
    _create_in_folder = serializers.CharField(required=False, allow_blank=True, write_only=True)

    class Meta:
        model = Survey
        fields = [
            "id",
            "name",
            "description",
            "type",
            "schedule",
            "linked_flag",
            "linked_flag_id",
            "linked_insight_id",
            "targeting_flag_id",
            "targeting_flag",
            "internal_targeting_flag",
            "targeting_flag_filters",
            "remove_targeting_flag",
            "questions",
            "conditions",
            "appearance",
            "created_at",
            "created_by",
            "start_date",
            "end_date",
            "archived",
            "responses_limit",
            "iteration_count",
            "iteration_frequency_days",
            "iteration_start_dates",
            "current_iteration",
            "current_iteration_start_date",
            "response_sampling_start_date",
            "response_sampling_interval_type",
            "response_sampling_interval",
            "response_sampling_limit",
            "response_sampling_daily_limits",
            "enable_partial_responses",
            "enable_iframe_embedding",
            "base_language",
            "translations",
            "_create_in_folder",
            "form_content",
        ]
        read_only_fields = ["id", "linked_flag", "targeting_flag", "created_at"]

    def to_representation(self, instance):
        data = super().to_representation(instance)
        data["conditions"] = get_survey_conditions_with_actions(instance)
        return data

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

        if not isinstance(value, dict):
            raise serializers.ValidationError("Appearance must be an object")

        thank_you_message = value.get("thankYouMessageHeader")
        if thank_you_message and nh3.is_html(thank_you_message):
            value["thankYouMessageHeader"] = nh3_clean_with_allow_list(thank_you_message)

        thank_you_description = value.get("thankYouMessageDescription")
        if thank_you_description and nh3.is_html(thank_you_description):
            value["thankYouMessageDescription"] = nh3_clean_with_allow_list(thank_you_description)

        thank_you_close_button = value.get("thankYouMessageCloseButtonText")
        if thank_you_close_button and nh3.is_html(thank_you_close_button):
            value["thankYouMessageCloseButtonText"] = nh3_clean_with_allow_list(thank_you_close_button)

        thank_you_description_content_type = value.get("thankYouMessageDescriptionContentType")
        if thank_you_description_content_type and thank_you_description_content_type not in ["text", "html"]:
            raise serializers.ValidationError("thankYouMessageDescriptionContentType must be one of ['text', 'html']")

        survey_popup_delay_seconds = value.get("surveyPopupDelaySeconds")
        if survey_popup_delay_seconds and survey_popup_delay_seconds < 0:
            raise serializers.ValidationError("Survey popup delay seconds must be a positive integer")

        survey_white_label = value.get("whiteLabel")
        if survey_white_label is not None and not isinstance(survey_white_label, bool):
            raise serializers.ValidationError("whiteLabel must be a boolean")

        # Check if the organization has the white labelling feature available
        use_survey_white_labelling = self.context["get_organization"]().is_feature_available(
            AvailableFeature.WHITE_LABELLING
        )

        if survey_white_label and not use_survey_white_labelling:
            raise serializers.ValidationError("You need to upgrade to Insights Enterprise to use white labelling")

        return value

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

        if not isinstance(value, dict):
            raise serializers.ValidationError("Conditions must be an object")

        return value

    def _resolve_base_language(self) -> str:
        """Resolve the survey's base language from incoming data, falling back to the instance, then 'en'."""
        initial_data = getattr(self, "initial_data", None)
        if isinstance(initial_data, dict) and "base_language" in initial_data:
            candidate = initial_data.get("base_language")
            if isinstance(candidate, str) and candidate.strip():
                return _normalize_language_code(candidate)
        if self.instance is not None:
            instance_lang = getattr(self.instance, "base_language", None)
            if isinstance(instance_lang, str) and instance_lang.strip():
                return _normalize_language_code(instance_lang)
        return DEFAULT_BASE_LANGUAGE

    def _validate_translation_language_key(self, raw_key: Any, base_language: str, context: str) -> str:
        """Validate and normalize a single translation language key. Raises ValidationError on bad input."""
        if not isinstance(raw_key, str) or not raw_key.strip():
            raise serializers.ValidationError(f"{context}: translation language code cannot be empty.")
        normalized = _normalize_language_code(raw_key)
        if normalized in REJECTED_TRANSLATION_KEYS:
            raise serializers.ValidationError(
                f"{context}: '{raw_key}' is not a valid translation language. "
                "The untranslated text is the original — set the survey's base language instead of adding it as a translation."
            )
        if not BCP47_LANGUAGE_CODE_RE.match(normalized):
            raise serializers.ValidationError(
                f"{context}: '{raw_key}' is not a valid language code. "
                "Use BCP-47 codes like 'en', 'es', or 'es-MX' (not language names like 'english')."
            )
        if normalized == base_language:
            raise serializers.ValidationError(
                f"{context}: '{raw_key}' matches the survey's base language ('{base_language}'). "
                "The original text already covers this language — translate to a different language."
            )
        return normalized

    def validate_base_language(self, value: Any) -> str:
        """Ensure base_language is a canonical BCP-47-ish code."""
        if value is None or (isinstance(value, str) and not value.strip()):
            return DEFAULT_BASE_LANGUAGE
        if not isinstance(value, str):
            raise serializers.ValidationError("base_language must be a string.")
        normalized = _normalize_language_code(value)
        if normalized in REJECTED_TRANSLATION_KEYS:
            raise serializers.ValidationError(
                f"'{value}' is not a valid language code. Use a BCP-47 code like 'en' or 'es'."
            )
        if not BCP47_LANGUAGE_CODE_RE.match(normalized):
            raise serializers.ValidationError(
                f"'{value}' is not a valid language code. Use BCP-47 codes like 'en', 'es', or 'es-MX'."
            )
        return normalized

    def validate_translations(self, value: Any) -> Optional[dict[str, dict[str, str]]]:
        """Validate survey-level translations.

        Write-strict, read-tolerant: pre-existing entries with unchanged content
        are passed through even if their language code is no longer valid, so users
        can keep saving surveys with legacy keys without losing data. The UI
        surfaces those entries with a fix-or-remove prompt.
        """
        if value is None:
            return value

        if not isinstance(value, dict):
            raise serializers.ValidationError("Translations must be an object")

        base_language = self._resolve_base_language()
        existing_translations: dict[str, Any] = {}
        if self.instance is not None:
            instance_translations = getattr(self.instance, "translations", None)
            if isinstance(instance_translations, dict):
                existing_translations = instance_translations

        cleaned_translations: dict[str, dict[str, str]] = {}
        for raw_lang_code, translation_data in value.items():
            if not isinstance(translation_data, dict):
                raise serializers.ValidationError(f"Translation for '{raw_lang_code}' must be an object")

            cleaned_translation: dict[str, str] = {}
            for field in [
                "name",
                "description",
                "thankYouMessageHeader",
                "thankYouMessageDescription",
                "thankYouMessageCloseButtonText",
            ]:
                if field in translation_data:
                    if not isinstance(translation_data[field], str):
                        raise serializers.ValidationError(
                            f"Translation for '{raw_lang_code}': '{field}' must be a string"
                        )
                    if nh3.is_html(translation_data[field]):
                        cleaned_translation[field] = nh3_clean_with_allow_list(translation_data[field])
                    else:
                        cleaned_translation[field] = translation_data[field]

            if not cleaned_translation:
                continue

            # Grandfather pre-existing keys: surveys saved before this validation keep saving, even when
            # their content is edited. New bad keys are still rejected below. The dashboard surfaces
            # these under "Legacy translation keys" so owners can clean up at their pace.
            if raw_lang_code in existing_translations:
                cleaned_translations[raw_lang_code] = cleaned_translation
                continue

            normalized_key = self._validate_translation_language_key(
                raw_lang_code, base_language, context="Survey translation"
            )
            # Detect collisions across normalization (e.g., legacy "EN" + new "en" both targeting "en").
            if any(_normalize_language_code(existing_key) == normalized_key for existing_key in cleaned_translations):
                raise serializers.ValidationError(
                    f"Survey translation: '{raw_lang_code}' collides with another translation for '{normalized_key}'."
                )
            cleaned_translations[normalized_key] = cleaned_translation

        return cleaned_translations

    def _validate_question_translations(
        self,
        translations_dict: Any,
        question_index: int,
        base_language: Optional[str] = None,
        existing_question_translations: Optional[dict[str, Any]] = None,
    ) -> dict[str, dict[str, Any]]:
        """Validate and sanitize translations for a single question.

        Write-strict, read-tolerant: pre-existing entries with unchanged content
        are passed through even if their language code is no longer valid.
        """
        # Use question_index + 1 for user-facing error messages
        question_num = question_index + 1

        if not isinstance(translations_dict, dict):
            raise serializers.ValidationError(f"Question {question_num}: translations must be an object")

        resolved_base_language = base_language or self._resolve_base_language()
        existing = existing_question_translations if isinstance(existing_question_translations, dict) else {}

        cleaned_translations: dict[str, dict[str, Any]] = {}

        for raw_lang_code, translation_data in translations_dict.items():
            if not isinstance(translation_data, dict):
                raise serializers.ValidationError(
                    f"Question {question_num}: Translation for '{raw_lang_code}' must be an object"
                )

            cleaned_translation: dict[str, Any] = {}

            # Validate and sanitize all translatable fields
            for field in ["question", "description", "buttonText", "lowerBoundLabel", "upperBoundLabel"]:
                if field in translation_data:
                    if not isinstance(translation_data[field], str):
                        raise serializers.ValidationError(
                            f"Question {question_num}: Translation '{raw_lang_code}' field '{field}' must be a string"
                        )
                    if nh3.is_html(translation_data[field]):
                        cleaned_translation[field] = nh3_clean_with_allow_list(translation_data[field])
                    else:
                        cleaned_translation[field] = translation_data[field]

            # Empty link is treated as absent rather than rejected, mirroring the base question.
            link = translation_data.get("link")
            if link:
                if not isinstance(link, str):
                    raise serializers.ValidationError(
                        f"Question {question_num}: Translation '{raw_lang_code}' field 'link' must be a string"
                    )
                cleaned_translation["link"] = self._validate_and_sanitize_link(link)

            # Validate and sanitize choices array
            if "choices" in translation_data:
                if not isinstance(translation_data["choices"], list):
                    raise serializers.ValidationError(
                        f"Question {question_num}: Translation '{raw_lang_code}' field 'choices' must be a list of strings"
                    )
                cleaned_translation["choices"] = self._validate_and_sanitize_choices(translation_data["choices"])

            if not cleaned_translation:
                continue

            # Grandfather pre-existing keys so edits to legacy translations still save (see validate_translations).
            if raw_lang_code in existing:
                cleaned_translations[raw_lang_code] = cleaned_translation
                continue

            normalized_key = self._validate_translation_language_key(
                raw_lang_code, resolved_base_language, context=f"Question {question_num} translation"
            )
            # Detect collisions across normalization (e.g., legacy "EN" + new "en" both targeting "en").
            if any(_normalize_language_code(existing_key) == normalized_key for existing_key in cleaned_translations):
                raise serializers.ValidationError(
                    f"Question {question_num} translation: '{raw_lang_code}' collides with another translation "
                    f"for '{normalized_key}'."
                )
            cleaned_translations[normalized_key] = cleaned_translation

        return cleaned_translations

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

        if not isinstance(value, list):
            raise serializers.ValidationError("Questions must be a list of objects")

        base_language = self._resolve_base_language()
        existing_questions: list[dict[str, Any]] = []
        if self.instance is not None:
            existing_value = getattr(self.instance, "questions", None)
            if isinstance(existing_value, list):
                existing_questions = [q for q in existing_value if isinstance(q, dict)]

        cleaned_questions = []
        for index, raw_question in enumerate(value):
            if not isinstance(raw_question, dict):
                raise serializers.ValidationError("Questions must be a list of objects")

            cleaned_question = {**raw_question}

            question_text = raw_question.get("question")

            if not question_text:
                raise serializers.ValidationError("Question text is required")

            if not isinstance(question_text, str):
                raise serializers.ValidationError("Question text must be a string")

            # Sanitize and assign question text
            if nh3.is_html(question_text):
                cleaned_question["question"] = nh3_clean_with_allow_list(question_text)
            else:
                cleaned_question["question"] = question_text

            description = raw_question.get("description")
            if description:
                if not isinstance(description, str):
                    raise serializers.ValidationError("Question description must be a string")
                # Sanitize and assign description
                if nh3.is_html(description):
                    cleaned_question["description"] = nh3_clean_with_allow_list(description)
                else:
                    cleaned_question["description"] = description

            for field_name, error_message in [
                ("buttonText", "Question buttonText must be a string"),
                ("lowerBoundLabel", "Question lowerBoundLabel must be a string"),
                ("upperBoundLabel", "Question upperBoundLabel must be a string"),
            ]:
                field_value = raw_question.get(field_name)
                if field_value:
                    if not isinstance(field_value, str):
                        raise serializers.ValidationError(error_message)
                    if nh3.is_html(field_value):
                        cleaned_question[field_name] = nh3_clean_with_allow_list(field_value)
                    else:
                        cleaned_question[field_name] = field_value

            # Validate choices first before translation validation to provide clearer error messages
            choices = raw_question.get("choices")
            if choices:
                if not isinstance(choices, list):
                    raise serializers.ValidationError("Question choices must be a list of strings")
                cleaned_question["choices"] = self._validate_and_sanitize_choices(choices)

            description_content_type = raw_question.get("descriptionContentType")
            if description_content_type and description_content_type not in ["text", "html"]:
                raise serializers.ValidationError("Question descriptionContentType must be one of ['text', 'html']")

            # Validate and sanitize inline translations
            if "translations" in raw_question:
                existing_question_translations: dict[str, Any] = {}
                if index < len(existing_questions):
                    candidate = existing_questions[index].get("translations")
                    if isinstance(candidate, dict):
                        existing_question_translations = candidate
                cleaned_translations = self._validate_question_translations(
                    raw_question["translations"],
                    index,
                    base_language=base_language,
                    existing_question_translations=existing_question_translations,
                )

                # Validate choices array length matches if present
                original_choices = cleaned_question.get("choices")
                for lang_code, translation_data in cleaned_translations.items():
                    if "choices" in translation_data:
                        if "choices" not in cleaned_question or not isinstance(original_choices, list):
                            translation_data.pop("choices", None)
                            continue

                        translated_choices = translation_data["choices"]
                        if len(translated_choices) != len(original_choices):
                            raise serializers.ValidationError(
                                f"Question {index + 1}: Translation '{lang_code}' has {len(translated_choices)} choices "
                                f"but question has {len(original_choices)} choices. Array lengths must match to avoid partial translations."
                            )

                cleaned_question["translations"] = cleaned_translations

            link = raw_question.get("link")
            if link:
                if not isinstance(link, str):
                    raise serializers.ValidationError(f"Question {index + 1}: link must be a string")
                cleaned_question["link"] = self._validate_and_sanitize_link(link)

            # Validate validation rules for open text questions
            validation_rules = raw_question.get("validation")
            if validation_rules is not None:
                if not isinstance(validation_rules, list):
                    raise serializers.ValidationError("Question validation must be a list of validation rules")

                valid_types = ["min_length", "max_length", "email"]
                for rule in validation_rules:
                    if not isinstance(rule, dict):
                        raise serializers.ValidationError("Each validation rule must be an object")

                    rule_type = rule.get("type")
                    if rule_type not in valid_types:
                        raise serializers.ValidationError(
                            f"Validation rule type must be one of: {', '.join(valid_types)}"
                        )

                    # Validate value for length rules
                    if rule_type in ["min_length", "max_length"]:
                        rule_value = rule.get("value")
                        if rule_value is not None:
                            if not isinstance(rule_value, int) or rule_value <= 0:
                                raise serializers.ValidationError(
                                    f"Validation rule value for {rule_type} must be a positive integer"
                                )

                # Cross-validate min_length <= max_length
                min_rule = next((r for r in validation_rules if r.get("type") == "min_length"), None)
                max_rule = next((r for r in validation_rules if r.get("type") == "max_length"), None)
                if min_rule and max_rule:
                    min_val = min_rule.get("value")
                    max_val = max_rule.get("value")
                    if min_val is not None and max_val is not None and min_val > max_val:
                        raise serializers.ValidationError("Minimum length cannot be greater than maximum length")

            cleaned_questions.append(cleaned_question)

        return cleaned_questions

    def validate_schedule(self, value):
        if value is not None and value not in ["once", "recurring", "always"]:
            raise serializers.ValidationError("Schedule must be one of: once, recurring, always")
        return value

    def validate(self, data):
        linked_flag_id = data.get("linked_flag_id")
        linked_flag = None
        if linked_flag_id:
            try:
                linked_flag = FeatureFlag.objects.get(pk=linked_flag_id, team_id=self.context["team_id"])
            except FeatureFlag.DoesNotExist:
                raise serializers.ValidationError("Feature Flag with this ID does not exist")

        targeting_flag_id = data.get("targeting_flag_id")
        if targeting_flag_id:
            try:
                FeatureFlag.objects.get(pk=targeting_flag_id, team_id=self.context["team_id"])
            except FeatureFlag.DoesNotExist:
                raise serializers.ValidationError("Targeting Feature Flag with this ID does not exist")

        linked_insight_id = data.get("linked_insight_id")
        if linked_insight_id:
            try:
                Insight.objects.get(pk=linked_insight_id, team_id=self.context["team_id"])
            except Insight.DoesNotExist:
                raise serializers.ValidationError("Insight with this ID does not exist")

        # Validate linkedFlagVariant if provided
        conditions = data.get("conditions") or {}
        linked_flag_variant = conditions.get("linkedFlagVariant")
        if linked_flag_variant and linked_flag and linked_flag_variant != "any":
            # Get available variants from the linked feature flag
            available_variants = [variant["key"] for variant in linked_flag.variants]
            if linked_flag_variant not in available_variants:
                if available_variants:
                    raise serializers.ValidationError(
                        f"Feature flag variant '{linked_flag_variant}' does not exist. Available variants: {', '.join(available_variants)}"
                    )
                else:
                    raise serializers.ValidationError(
                        f"Feature flag variant '{linked_flag_variant}' specified but the linked feature flag has no variants"
                    )
        elif linked_flag_variant and not linked_flag_id:
            raise serializers.ValidationError("linkedFlagVariant can only be used when a linked_flag_id is specified")

        if (
            self.context["request"].method == "POST"
            and Survey.objects.filter(name=data.get("name"), team__project_id=self.context["project_id"]).exists()
        ):
            raise serializers.ValidationError("There is already a survey with this name.", code="unique")

        existing_survey: Survey | None = self.instance

        if (
            existing_survey
            and existing_survey.name != data.get("name")
            and Survey.objects.filter(name=data.get("name"), team__project_id=self.context["project_id"])
            .exclude(id=existing_survey.id)
            .exists()
        ):
            raise serializers.ValidationError("There is already another survey with this name.", code="unique")

        if data.get("targeting_flag_filters"):
            groups = (data.get("targeting_flag_filters") or {}).get("groups") or []
            full_rollout = any(
                group.get("rollout_percentage") in [100, None] and len(group.get("properties", [])) == 0
                for group in groups
            )

            if full_rollout:
                raise serializers.ValidationError(
                    "Invalid operation: User targeting rolls out to everyone. If you want to roll out to everyone, delete this targeting",
                    code="invalid",
                )

        response_sampling_start_date = data.get("response_sampling_start_date")
        if response_sampling_start_date is not None:
            today_utc = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
            if response_sampling_start_date < today_utc:
                raise serializers.ValidationError(
                    {
                        "response_sampling_start_date": f"Response sampling start date must be today or a future date in UTC. Got {response_sampling_start_date} when current time is {today_utc}"
                    }
                )

        response_sampling_interval = data.get("response_sampling_interval")
        if response_sampling_interval is not None and response_sampling_interval <= 0:
            raise serializers.ValidationError(
                {"response_sampling_interval": "Response sampling interval must be greater than 0."}
            )

        response_sampling_limit = data.get("response_sampling_limit", 0)
        if (
            response_sampling_limit is not None
            and response_sampling_limit > 0
            and response_sampling_interval > 0
            and response_sampling_start_date is None
        ):
            raise serializers.ValidationError(
                {
                    "response_sampling_start_date": "Response sampling start date should be set if response_sampling_start_date is not zero."
                }
            )

        # Validate that question shuffling and branching are not used together
        # Check both incoming data and existing survey data (existing_survey is None for POST)
        appearance = data.get("appearance")
        questions = data.get("questions")

        effective_appearance = {}
        if existing_survey and existing_survey.appearance:
            effective_appearance = dict(existing_survey.appearance)
        if appearance is not None:
            effective_appearance.update(appearance)

        effective_questions = (
            questions if questions is not None else (existing_survey.questions if existing_survey else [])
        )

        shuffle_questions = effective_appearance.get("shuffleQuestions", False)

        if shuffle_questions and effective_questions:
            has_branching = any(question.get("branching") is not None for question in effective_questions)

            if has_branching:
                raise serializers.ValidationError(
                    "Question shuffling and question branching cannot be used together. "
                    "Please disable one of these features."
                )

        # Validate external survey constraints
        if data.get("type") == Survey.SurveyType.EXTERNAL_SURVEY:
            errors = {}

            # Check prohibited fields
            for field in FIELDS_NOT_APPLICABLE_TO_EXTERNAL_SURVEYS:
                if data.get(field) is not None:
                    errors[field] = f"{field} is not allowed for external surveys"

            # Check prohibited condition fields
            if data.get("conditions"):
                condition_errors = []
                conditions = data["conditions"]

                # Check individual condition fields
                for field in CONDITION_FIELDS_NOT_APPLICABLE_TO_EXTERNAL_SURVEYS:
                    if field in conditions and conditions[field] is not None:
                        condition_errors.append(field)

                # Check actions/events if they have values
                for field in ["actions", "events"]:
                    if field in conditions and isinstance(conditions[field], dict):
                        values = conditions[field].get("values", [])
                        if values:
                            condition_errors.append(field)

                if condition_errors:
                    errors["conditions"] = (
                        f"The following condition fields are not allowed for external surveys: {', '.join(condition_errors)}"
                    )

            # Check prohibited appearance fields
            if "appearance" in data and data["appearance"] and "surveyPopupDelaySeconds" in data["appearance"]:
                if data["appearance"]["surveyPopupDelaySeconds"] is not None:
                    errors["appearance"] = "surveyPopupDelaySeconds is not allowed for external surveys"

            if errors:
                raise serializers.ValidationError(errors)

        # Cross-field: base_language must not collide with any existing translation key.
        # validate_translations already covers the "translations in payload" case; this catches partial
        # updates where only base_language is sent (otherwise get_survey_api_translations would silently
        # drop the matching translation from the SDK payload, making it unreachable with no warning).
        incoming_base_language = data.get("base_language")
        if incoming_base_language and "translations" not in data and self.instance is not None:
            normalized_base = _normalize_language_code(incoming_base_language)
            instance_translations = getattr(self.instance, "translations", None) or {}
            for existing_key in instance_translations:
                if isinstance(existing_key, str) and _normalize_language_code(existing_key) == normalized_base:
                    raise serializers.ValidationError(
                        {
                            "base_language": (
                                f"'{incoming_base_language}' already exists as a translation key on this survey. "
                                "Remove the translation first, or pick a different base language."
                            )
                        }
                    )

        return data

    def create(self, validated_data):
        if "remove_targeting_flag" in validated_data:
            validated_data.pop("remove_targeting_flag")

        validated_data["team_id"] = self.context["team_id"]
        if validated_data.get("targeting_flag_id"):
            # Attaching an existing flag takes write control of it: its active state is later
            # synced to the survey's lifecycle (start/stop/archive). Require feature_flag:write
            # so survey:write alone can't point a survey at an arbitrary flag and toggle it.
            assert_feature_flag_write_scope(
                self.context["request"],
                action="survey.create.targeting_flag_id",
                resource_scope="survey:write",
                team_id=self.context["team_id"],
                feature_flag_id=validated_data["targeting_flag_id"],
            )
        if validated_data.get("targeting_flag_filters"):
            assert_feature_flag_write_scope(
                self.context["request"],
                action="survey.create",
                resource_scope="survey:write",
                team_id=self.context["team_id"],
            )
            targeting_feature_flag = self._create_or_update_targeting_flag(
                None, validated_data["targeting_flag_filters"], validated_data["name"]
            )
            validated_data["targeting_flag_id"] = targeting_feature_flag.id
            validated_data.pop("targeting_flag_filters")

        if "targeting_flag_filters" in validated_data:
            validated_data.pop("targeting_flag_filters")

        validated_data["created_by"] = self.context["request"].user
        instance = super().create(validated_data)
        self._add_user_survey_interacted_filters(instance)
        self._associate_actions(instance, validated_data.get("conditions"))
        self._add_internal_response_sampling_filters(instance)

        team = Team.objects.get(id=self.context["team_id"])
        log_activity(
            organization_id=team.organization_id,
            team_id=self.context["team_id"],
            user=self.context["request"].user,
            was_impersonated=is_impersonated(self.context["request"]),
            item_id=instance.id,
            scope="Survey",
            activity="created",
            detail=Detail(name=instance.name),
        )

        return instance

    def update(self, instance: Survey, validated_data):
        before_update = Survey.objects.get(pk=instance.pk)
        user = self.context["request"].user
        changes = []

        if validated_data.get("targeting_flag_id"):
            # Re-pointing the survey at an existing flag takes write control of it (the
            # lifecycle sync below can toggle its active state), so require feature_flag:write.
            assert_feature_flag_write_scope(
                self.context["request"],
                action="survey.update.targeting_flag_id",
                resource_scope="survey:write",
                team_id=self.context["team_id"],
                feature_flag_id=validated_data["targeting_flag_id"],
            )

        if validated_data.get("remove_targeting_flag"):
            if instance.targeting_flag:
                assert_feature_flag_write_scope(
                    self.context["request"],
                    action="survey.update.remove_targeting_flag",
                    resource_scope="survey:write",
                    team_id=self.context["team_id"],
                    feature_flag_id=instance.targeting_flag_id,
                )
                # Manually delete the flag and log the change
                # The `changes_between` method won't catch this because the flag (and underlying ForeignKey relationship)
                # will have been deleted by the time the `changes_between` method is called, so we need to log the change manually
                changes.append(
                    Change(type="Survey", field="targeting_flag", action="deleted", before=instance.targeting_flag)
                )
                instance.targeting_flag.delete()
                validated_data["targeting_flag_id"] = None
            validated_data.pop("remove_targeting_flag")

            # make sure instance.targeting_flag is gone
            instance.refresh_from_db()

        # if the target flag filters come back with data, update the targeting feature flag if there is one, otherwise create a new one
        if validated_data.get("targeting_flag_filters"):
            assert_feature_flag_write_scope(
                self.context["request"],
                action="survey.update.targeting_flag_filters",
                resource_scope="survey:write",
                team_id=self.context["team_id"],
                feature_flag_id=instance.targeting_flag_id,
            )
            new_filters = validated_data["targeting_flag_filters"]
            if instance.targeting_flag:
                existing_targeting_flag = instance.targeting_flag
                existing_targeting_flag_filters = existing_targeting_flag.filters
                serialized_data_filters = {
                    **existing_targeting_flag_filters,
                    **new_filters,
                }
                # Log the existing filter change
                # The `changes_between` method won't catch this because the flag (and underlying ForeignKey relationship)
                # will have been deleted by the time the `changes_between` method is called, so we need to log the change manually
                changes.append(
                    Change(
                        type="Survey",
                        field="targeting_flag_filters",
                        action="changed",
                        before=existing_targeting_flag_filters,
                        after=new_filters,
                    )
                )
                self._create_or_update_targeting_flag(instance.targeting_flag, serialized_data_filters)
            else:
                new_flag = self._create_or_update_targeting_flag(
                    None, new_filters, instance.name, bool(instance.start_date)
                )
                # Log the new filter change
                # The `changes_between` method won't catch this because the flag (and underlying ForeignKey relationship)
                # will have been deleted by the time the `changes_between` method is called, so we need to log the change manually
                changes.append(
                    Change(type="Survey", field="targeting_flag_filters", action="created", after=new_filters)
                )
                validated_data["targeting_flag_id"] = new_flag.id
            validated_data.pop("targeting_flag_filters")

        iteration_count = validated_data.get("iteration_count", None)
        if (
            instance.current_iteration is not None
            and iteration_count is not None
            and instance.current_iteration > iteration_count > 0
        ):
            raise serializers.ValidationError(
                f"Cannot change survey recurrence to {iteration_count}, should be at least {instance.current_iteration}"
            )

        if iteration_count is not None:
            instance.iteration_count = iteration_count
            instance.iteration_frequency_days = validated_data.get("iteration_frequency_days")

        instance = super().update(instance, validated_data)

        team = Team.objects.get(id=self.context["team_id"])
        # `changes_between` will not catch changes to the ForeignKey relationships
        # so it's useful for any changes to the Survey model itself, but not for the related models
        non_foreign_table_relation_changes = changes_between(
            "Survey",
            previous=before_update,
            current=instance,
        )
        changes.extend(non_foreign_table_relation_changes)
        log_activity(
            organization_id=team.organization_id,
            team_id=self.context["team_id"],
            user=self.context["request"].user,
            was_impersonated=is_impersonated(self.context["request"]),
            item_id=instance.id,
            scope="Survey",
            activity="updated",
            detail=Detail(changes=changes, name=instance.name),
        )

        # Report survey events based on start_date and end_date changes

        properties = {
            "name": instance.name,
            "id": instance.id,
            "survey_type": instance.type,
            "question_types": [question.get("type") for question in instance.questions] if instance.questions else [],
            "created_at": instance.created_at,
            "start_date": instance.start_date,
            "end_date": instance.end_date,
        }
        if before_update.start_date is None and instance.start_date is not None:
            report_user_action(
                user,
                "survey launched",
                properties,
                team=team,
                request=self.context["request"],
            )
        elif before_update.end_date is None and instance.end_date is not None:
            report_user_action(
                user,
                "survey stopped",
                properties,
                team=team,
                request=self.context["request"],
            )
        elif before_update.start_date is not None and before_update.end_date is not None and instance.end_date is None:
            report_user_action(
                user,
                "survey resumed",
                properties,
                team=team,
                request=self.context["request"],
            )

        should_flag_be_active = self._should_survey_flags_be_active(instance)

        # Intentionally NOT gated on feature_flag:write: this only syncs the active state of
        # the survey's OWN dedicated targeting flag to the survey's running state (start/stop/
        # archive), which survey:write already controls. The flag's definition (its filters)
        # can only be set via the gated targeting_flag_filters path. Enforcement covers
        # definition writes, not this lifecycle mirror.
        if instance.targeting_flag:
            instance.targeting_flag.active = should_flag_be_active
            instance.targeting_flag.save()

        self._add_user_survey_interacted_filters(instance)
        self._associate_actions(instance, validated_data.get("conditions"))
        self._add_internal_response_sampling_filters(instance)
        return instance

    def _add_internal_response_sampling_filters(self, instance: Survey):
        if instance.response_sampling_daily_limits is None:
            return
        if instance.internal_response_sampling_flag is not None:
            return

        sampling_filters = {
            "groups": [
                {
                    "variant": "",
                    "rollout_percentage": 100,
                    "properties": [],
                }
            ]
        }

        instance.internal_response_sampling_flag = self._create_or_update_targeting_flag(
            None, sampling_filters, instance.name, bool(instance.start_date), flag_name_suffix="-sampling"
        )
        instance.save()

    def _associate_actions(self, instance: Survey, conditions):
        if conditions is None:
            instance.actions.clear()
            return

        actions = conditions.get("actions")
        if actions is None:
            instance.actions.clear()
            return

        values = actions.get("values")
        if values is None or len(values) == 0:
            instance.actions.clear()
            return

        action_ids = (value.get("id") for value in values)

        instance.actions.set(Action.objects.filter(team__project_id=self.context["project_id"], id__in=action_ids))
        instance.save()

    def _should_survey_flags_be_active(self, instance: Survey) -> bool:
        return bool(instance.start_date) and not instance.end_date and not instance.archived

    def _add_user_survey_interacted_filters(self, instance: Survey):
        should_flag_be_active = self._should_survey_flags_be_active(instance)

        survey_key = f"{instance.id}"
        if instance.iteration_count is not None and instance.iteration_count > 0:
            survey_key = f"{instance.id}/{instance.current_iteration or 1}"

        base_properties = [
            {
                "key": f"{SurveyEventProperties.SURVEY_DISMISSED}/{survey_key}",
                "value": "is_not_set",
                "operator": "is_not_set",
                "type": "person",
            },
            {
                "key": f"{SurveyEventProperties.SURVEY_RESPONDED}/{survey_key}",
                "value": "is_not_set",
                "operator": "is_not_set",
                "type": "person",
            },
        ]

        wait_period_days = None
        if instance.conditions and isinstance(instance.conditions, dict):
            wait_period_days = instance.conditions.get("seenSurveyWaitPeriodInDays")

        if wait_period_days is not None and wait_period_days > 0:
            user_submitted_dismissed_filter = {
                "groups": [
                    {
                        "variant": "",
                        "rollout_percentage": 100,
                        "properties": [
                            *base_properties,
                            {
                                "key": SurveyEventProperties.SURVEY_LAST_SEEN_DATE,
                                "value": "is_not_set",
                                "operator": "is_not_set",
                                "type": "person",
                            },
                        ],
                    },
                    {
                        "variant": "",
                        "rollout_percentage": 100,
                        "properties": [
                            *base_properties,
                            {
                                "key": SurveyEventProperties.SURVEY_LAST_SEEN_DATE,
                                "value": f"{int(wait_period_days)}d",
                                "operator": "is_date_before",
                                "type": "person",
                            },
                        ],
                    },
                ]
            }
        else:
            user_submitted_dismissed_filter = {
                "groups": [
                    {
                        "variant": "",
                        "rollout_percentage": 100,
                        "properties": base_properties,
                    }
                ]
            }

        if instance.internal_targeting_flag:
            existing_targeting_flag = instance.internal_targeting_flag
            # Note: new filters must come LAST to overwrite old iteration-unaware properties
            serialized_data_filters = {**existing_targeting_flag.filters, **user_submitted_dismissed_filter}

            internal_targeting_flag = self._create_or_update_targeting_flag(
                instance.internal_targeting_flag, serialized_data_filters, flag_name_suffix="-custom"
            )

            internal_targeting_flag.active = should_flag_be_active
            internal_targeting_flag.save()

            instance.internal_targeting_flag_id = internal_targeting_flag.id

            instance.save()
        else:
            new_flag = self._create_or_update_targeting_flag(
                None,
                user_submitted_dismissed_filter,
                instance.name,
                should_flag_be_active,
                flag_name_suffix="-custom",
            )
            instance.internal_targeting_flag_id = new_flag.id
            instance.save()

    def _create_or_update_targeting_flag(
        self, existing_flag=None, filters=None, name=None, active=False, flag_name_suffix=None
    ):
        with create_flag_with_survey_errors():
            # Ensure the request method is set correctly for validation
            if existing_flag:
                self.context["request"].method = "PATCH"
                existing_flag_serializer = FeatureFlagSerializer(
                    existing_flag,
                    data={"filters": filters},
                    partial=True,
                    context=self.context,
                )
                existing_flag_serializer.is_valid(raise_exception=True)
                return existing_flag_serializer.save()
            else:
                self.context["request"].method = "POST"
                random_id = generate("1234567890abcdef", 10)
                feature_flag_key = slugify(f"{SURVEY_TARGETING_FLAG_PREFIX}{random_id}{flag_name_suffix or ''}")
                feature_flag_serializer = FeatureFlagSerializer(
                    data={
                        "key": feature_flag_key,
                        "name": f"Targeting flag for survey {name}",
                        "filters": filters,
                        "active": active,
                        "creation_context": "surveys",
                    },
                    context=self.context,
                )

                feature_flag_serializer.is_valid(raise_exception=True)
                return feature_flag_serializer.save()


class SurveySerializerCreateUpdateOnlySchema(SurveySerializerCreateUpdateOnly):
    linked_flag_id = serializers.IntegerField(
        required=False,
        write_only=True,
        allow_null=True,
        help_text="The feature flag linked to this survey.",
    )
    targeting_flag_id = serializers.IntegerField(
        required=False,
        write_only=True,
        help_text="An existing targeting flag to use for this survey.",
    )
    targeting_flag_filters = FeatureFlagFiltersSchemaSerializer(  # type: ignore[assignment]
        required=False,
        write_only=True,
        allow_null=True,
        help_text="Target specific users based on their properties. Example: {groups: [{properties: [{key: 'email', value: ['@company.com'], operator: 'icontains'}], rollout_percentage: 100}]}",
    )
    remove_targeting_flag = serializers.BooleanField(
        required=False,
        write_only=True,
        allow_null=True,
        help_text="Set to true to completely remove all targeting filters from the survey, making it visible to all users (subject to other display conditions like URL matching).",
    )
    questions = SurveyQuestionsSchemaField(
        child=serializers.DictField(),
        required=False,
        allow_null=True,
        help_text=Survey._meta.get_field("questions").help_text,
    )
    appearance = SurveyAppearanceSchemaSerializer(
        required=False,
        allow_null=True,
        help_text="Survey appearance customization.",
    )
    conditions = SurveyConditionsSchemaSerializer(
        required=False,
        allow_null=True,
        help_text="Display and targeting conditions for the survey.",
    )
    schedule = serializers.ChoiceField(  # type: ignore[assignment]
        choices=Survey.Schedule.choices,
        required=False,
        allow_null=True,
        help_text=(
            "Survey scheduling behavior: 'once' = show once per user (default), "
            "'recurring' = repeat based on iteration_count and iteration_frequency_days settings, "
            "'always' = show every time conditions are met (mainly for widget surveys)"
        ),
    )
    responses_limit = serializers.IntegerField(
        required=False,
        allow_null=True,
        help_text="The maximum number of responses before automatically stopping the survey.",
    )
    iteration_count = serializers.IntegerField(
        required=False,
        allow_null=True,
        max_value=MAX_ITERATION_COUNT,
        min_value=1,
        help_text="For a recurring schedule, this field specifies the number of times the survey should be shown to the user. Use 1 for 'once every X days', higher numbers for multiple repetitions. Works together with iteration_frequency_days to determine the overall survey schedule.",
    )
    iteration_frequency_days = serializers.IntegerField(
        required=False,
        allow_null=True,
        min_value=1,
        max_value=365,
        help_text="For a recurring schedule, this field specifies the interval in days between each survey instance shown to the user, used alongside iteration_count for precise scheduling.",
    )
    enable_partial_responses = serializers.BooleanField(
        required=False,
        allow_null=True,
        help_text="When at least one question is answered, the response is stored (true). The response is stored when all questions are answered (false).",
    )

    class Meta(SurveySerializerCreateUpdateOnly.Meta):
        extra_kwargs = {
            "name": {"help_text": "Survey name.", "min_length": 1},
            "description": {"help_text": "Survey description."},
            "type": {"help_text": "Survey type."},
            "start_date": {
                "help_text": "Setting this will launch the survey immediately. Don't add a start_date unless explicitly requested to do so."
            },
            "end_date": {
                "help_text": "When the survey stopped being shown to users. Setting this will complete the survey."
            },
            "archived": {"help_text": "Archive state for the survey."},
        }


class UUIDInFilter(django_filters.BaseInFilter, django_filters.UUIDFilter):
    pass


class SurveyFilterSet(FilterSet):
    ids = UUIDInFilter(
        field_name="id",
        label="Filter to a comma-separated list of survey IDs. IDs that don't exist are silently omitted rather than erroring.",
    )

    class Meta:
        model = Survey
        fields = ["archived", "type"]


@extend_schema_view(
    create=extend_schema(request=SurveySerializerCreateUpdateOnlySchema),
    partial_update=extend_schema(request=SurveySerializerCreateUpdateOnlySchema),
    list=extend_schema(
        parameters=[
            OpenApiParameter(
                name="search",
                type=OpenApiTypes.STR,
                description="Match against survey `name` and `description`. Returns exact (case-insensitive substring) matches only; if no exact match exists, returns similar (fuzzy trigram — typos, prefix-as-you-type) matches instead. Each result's `search_match_type` is `exact` or `similar`.",
            ),
        ],
    ),
)
class SurveyViewSet(TeamAndOrgViewSetMixin, AccessControlViewSetMixin, viewsets.ModelViewSet):
    scope_object = "survey"
    queryset = Survey.objects.select_related(
        "linked_flag", "linked_insight", "targeting_flag", "internal_targeting_flag"
    ).all()
    filter_backends = [DjangoFilterBackend]
    filterset_class = SurveyFilterSet

    def get_serializer_class(self) -> type[serializers.Serializer]:
        if self.request.method == "POST" or self.request.method == "PATCH":
            return SurveySerializerCreateUpdateOnly
        else:
            return SurveySerializer

    def safely_get_queryset(self, queryset):
        queryset = queryset.exclude(product_tour__isnull=False)
        if self.action == "list":
            search = self.request.GET.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."}
                    )
                # Search applies its own relevance ordering — don't override it.
                queryset = self._apply_search(queryset, search)
            else:
                # Newest first — stable order for pagination and surfaces recent surveys first in pickers.
                queryset = queryset.order_by("-created_at")
        return queryset

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

    @tracer.start_as_current_span("SurveyViewSet.list")
    def list(self, request: request.Request, *args: Any, **kwargs: Any) -> Response:
        response = super().list(request, *args, **kwargs)
        if request.query_params.get("search"):
            data = response.data if isinstance(response.data, dict) else {}
            results_len = data.get("count", len(data.get("results", [])))
            span = trace.get_current_span()
            span.set_attribute("survey.search.result_count", results_len)
            span.set_attribute("survey.search.empty", results_len == 0)
        return response

    @staticmethod
    @tracer.start_as_current_span("SurveyViewSet._apply_search")
    def _apply_search(queryset: QuerySet, search: str) -> QuerySet:
        return apply_trigram_search(
            queryset,
            search,
            span_prefix="survey.search",
            fields=(NAME_FIELD, DESCRIPTION_FIELD),
            tiebreakers=("name",),
        )

    def destroy(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        instance = self.get_object()
        related_targeting_flag = instance.targeting_flag
        related_internal_targeting_flag = instance.internal_targeting_flag
        # Only the user-defined targeting flag is gated; the internal targeting flag is a
        # Insights-managed implementation detail, so deleting a survey that only has one
        # must not require `feature_flag:write`.
        if related_targeting_flag:
            assert_feature_flag_write_scope(
                request,
                action="survey.destroy",
                resource_scope="survey:write",
                team_id=self.team_id,
                feature_flag_id=related_targeting_flag.id,
            )
        if related_targeting_flag:
            related_targeting_flag.delete()
        if related_internal_targeting_flag:
            related_internal_targeting_flag.delete()

        log_activity(
            organization_id=self.organization.id,
            team_id=self.team_id,
            user=cast(User, self.request.user),
            was_impersonated=is_impersonated(request),
            item_id=instance.id,
            scope="Survey",
            activity="deleted",
            detail=Detail(name=instance.name),
        )

        return super().destroy(request, *args, **kwargs)

    @extend_schema(
        operation_id="surveys_launch",
        description=(
            "Launch a survey by setting `start_date` to the current time. No-op if the survey is already launched "
            "(start_date set in the past) — returns the existing state unchanged. Does not affect archived surveys "
            "or surveys with an end_date in the past; unarchive or extend the end_date first."
        ),
        request=None,
        responses={200: SurveySerializer},
    )
    @action(methods=["POST"], detail=True, required_scopes=["survey:write"])
    def launch(self, request: request.Request, **kwargs: Any) -> Response:
        survey = self.get_object()
        now = datetime.now(UTC)
        if survey.archived:
            raise exceptions.ValidationError("Cannot launch an archived survey. Unarchive it first.")
        if survey.end_date and survey.end_date <= now:
            raise exceptions.ValidationError(
                "Cannot launch a survey with end_date in the past. Extend the end_date first."
            )
        if survey.start_date and survey.start_date <= now:
            # Already launched — no-op, return current state.
            return Response(SurveySerializer(survey, context=self.get_serializer_context()).data)

        previous_start = survey.start_date
        survey.start_date = now
        survey.save(update_fields=["start_date"])

        log_activity(
            organization_id=self.organization.id,
            team_id=self.team_id,
            user=cast(User, self.request.user),
            was_impersonated=is_impersonated(request),
            item_id=survey.id,
            scope="Survey",
            activity="launched",
            detail=Detail(
                name=survey.name,
                changes=[
                    Change(
                        type="Survey",
                        action="changed",
                        field="start_date",
                        before=previous_start,
                        after=survey.start_date,
                    )
                ],
            ),
        )

        return Response(SurveySerializer(survey, context=self.get_serializer_context()).data)

    @extend_schema(
        operation_id="surveys_stop",
        description=(
            "Stop a survey by setting `end_date` to the current time. No new responses are accepted after this; "
            "existing responses remain available. No-op if the survey already has an end_date in the past."
        ),
        request=None,
        responses={200: SurveySerializer},
    )
    @action(methods=["POST"], detail=True, required_scopes=["survey:write"])
    def stop(self, request: request.Request, **kwargs: Any) -> Response:
        survey = self.get_object()
        now = datetime.now(UTC)
        if survey.archived:
            raise exceptions.ValidationError("Cannot stop an archived survey. Unarchive it first if needed.")
        if survey.end_date and survey.end_date <= now:
            return Response(SurveySerializer(survey, context=self.get_serializer_context()).data)

        previous_end = survey.end_date
        survey.end_date = now
        survey.save(update_fields=["end_date"])

        log_activity(
            organization_id=self.organization.id,
            team_id=self.team_id,
            user=cast(User, self.request.user),
            was_impersonated=is_impersonated(request),
            item_id=survey.id,
            scope="Survey",
            activity="stopped",
            detail=Detail(
                name=survey.name,
                changes=[
                    Change(
                        type="Survey",
                        action="changed",
                        field="end_date",
                        before=previous_end,
                        after=survey.end_date,
                    )
                ],
            ),
        )

        return Response(SurveySerializer(survey, context=self.get_serializer_context()).data)

    def _get_partial_responses_filter(self, base_conditions_sql: builtins.list[str]) -> str:
        return partial_responses_filter(base_conditions_sql)

    def _get_archived_responses_filter(self, survey_id: str | None = None) -> tuple[str, dict]:
        return archived_responses_filter(survey_id, self.team_id)

    @action(methods=["GET"], detail=False, required_scopes=["survey:read"])
    def responses_count(self, request: request.Request, **kwargs):
        """Get response counts for all surveys.

        Args:
            exclude_archived: Optional boolean to exclude archived responses (default: false, includes archived)
            survey_ids: Optional comma-separated list of survey IDs to filter by

        Returns:
            Dictionary mapping survey IDs to response counts
        """
        exclude_archived = request.query_params.get("exclude_archived", "false").lower() == "true"
        survey_ids_param = request.query_params.get("survey_ids")

        earliest_survey_start_date = Survey.objects.filter(team__project_id=self.project_id).aggregate(
            Min("start_date")
        )["start_date__min"]

        if not earliest_survey_start_date:
            # If there are no surveys or none have a start date, there can be no responses.
            return Response({})

        params = {"team_id": self.team_id, "timestamp": earliest_survey_start_date}
        survey_id_expr = get_survey_property_string_expr(SurveyEventProperties.SURVEY_ID)

        partial_responses_filter = self._get_partial_responses_filter(
            base_conditions_sql=[
                "team_id = %(team_id)s",
                "timestamp >= %(timestamp)s",
            ],
        )

        archived_filter = ""
        if exclude_archived:
            archived_filter_sql, archived_params = self._get_archived_responses_filter()
            if archived_filter_sql:
                archived_filter = f"AND {archived_filter_sql}"
                params.update(archived_params)

        survey_ids_filter = ""
        if survey_ids_param:
            survey_ids = [sid.strip() for sid in survey_ids_param.split(",") if sid.strip()]
            if survey_ids:
                survey_ids_filter = f"AND {survey_id_expr} IN %(survey_ids)s"
                params["survey_ids"] = survey_ids

        query = f"""
            SELECT
                {survey_id_expr} as survey_id,
                count()
            FROM events
            WHERE
                team_id = %(team_id)s
                AND event = '{SurveyEventName.SENT}'
                AND timestamp >= %(timestamp)s
                AND {partial_responses_filter}
                {archived_filter}
                {survey_ids_filter}
            GROUP BY survey_id
        """

        tag_queries(product=ProductKey.SURVEYS, feature=Feature.QUERY)
        data = sync_execute(query, params)

        counts = {}
        for survey_id, count in data:
            counts[survey_id] = count

        return Response(counts)

    def _validate_and_parse_dates(
        self, date_from: str | None, date_to: str | None
    ) -> tuple[datetime | None, datetime | None]:
        try:
            return validate_and_parse_dates(date_from, date_to)
        except ValueError as exc:
            raise exceptions.ValidationError(str(exc)) from exc

    def _process_survey_results(
        self, results: builtins.list[tuple[str, int, int, datetime | None, datetime | None]]
    ) -> SurveyStats:
        return process_survey_results(results)

    def _calculate_rates(self, stats: SurveyStats) -> SurveyRates:
        return calculate_rates(stats)

    def _get_survey_stats(
        self, date_from: str | None, date_to: str | None, survey_id: str | None = None, exclude_archived: bool = False
    ) -> dict:
        """Get survey statistics from Datastore.

        Args:
            date_from: Optional ISO timestamp for start date with timezone info
            date_to: Optional ISO timestamp for end date with timezone info
            survey_id: Optional survey ID to filter for. If None, gets stats for all surveys.
            exclude_archived: If True, exclude archived responses. Defaults to False (includes archived).

        Returns:
            Dictionary containing survey statistics and rates
        """
        # Shared helper raises ValueError for bad dates; surface it as a REST 400 here.
        try:
            return get_survey_stats(
                team_id=self.team_id,
                date_from=date_from,
                date_to=date_to,
                survey_id=survey_id,
                exclude_archived=exclude_archived,
            )
        except ValueError as exc:
            raise exceptions.ValidationError(str(exc)) from exc

    @extend_schema(
        parameters=[
            OpenApiParameter(
                "date_from",
                OpenApiTypes.DATETIME,
                required=False,
                description="Optional ISO timestamp for start date (e.g. 2024-01-01T00:00:00Z)",
            ),
            OpenApiParameter(
                "date_to",
                OpenApiTypes.DATETIME,
                required=False,
                description="Optional ISO timestamp for end date (e.g. 2024-01-31T23:59:59Z)",
            ),
            OpenApiParameter(
                "include_per_question_stats",
                OpenApiTypes.BOOL,
                required=False,
                description=(
                    "When true, also return per-question response counts and answer distributions. "
                    "Adds one extra InsightsQL query per question, so leave off unless you need the breakdown."
                ),
            ),
        ],
        responses={
            200: inline_serializer(
                name="SurveyStatsResponse",
                fields={
                    "survey_id": serializers.CharField(help_text="The survey ID these stats belong to."),
                    "start_date": serializers.DateTimeField(
                        allow_null=True, help_text="When the survey started collecting responses."
                    ),
                    "end_date": serializers.DateTimeField(
                        allow_null=True, help_text="When the survey stopped collecting responses."
                    ),
                    "stats": serializers.DictField(
                        help_text="Event counts keyed by event name (survey shown, survey dismissed, survey sent)."
                    ),
                    "rates": serializers.DictField(help_text="Calculated response and dismissal rates."),
                    "per_question_stats": serializers.ListField(
                        required=False,
                        help_text=(
                            "Per-question response counts and distributions. Only present when "
                            "include_per_question_stats=true was passed. For rating questions includes `average`; "
                            "for choice/rating questions `distribution` maps answer value to count; for open "
                            "questions `distribution` is empty (use surveys-responses-list to read free-text)."
                        ),
                    ),
                },
            )
        },
    )
    @action(methods=["GET"], detail=True, url_path="stats", required_scopes=["survey:read"])
    def survey_stats(self, request: request.Request, **kwargs) -> Response:
        """Get survey response statistics for a specific survey.

        Args:
            date_from: Optional ISO timestamp for start date (e.g. 2024-01-01T00:00:00Z)
            date_to: Optional ISO timestamp for end date (e.g. 2024-01-31T23:59:59Z)
            exclude_archived: Optional boolean to exclude archived responses (default: false, includes archived)
            include_per_question_stats: Optional boolean to include per-question response counts and distributions

        Returns:
            Survey statistics including event counts, unique respondents, and conversion rates
        """
        survey_id = kwargs["pk"]
        date_from = request.query_params.get("date_from", None)
        date_to = request.query_params.get("date_to", None)
        exclude_archived = request.query_params.get("exclude_archived", "false").lower() == "true"
        include_per_question_stats = request.query_params.get("include_per_question_stats", "false").lower() == "true"

        try:
            survey = self.get_object()
        except Survey.DoesNotExist:
            raise exceptions.NotFound("Survey not found")

        response_data = self._get_survey_stats(date_from, date_to, survey_id, exclude_archived)

        # Add survey metadata
        response_data["survey_id"] = survey_id
        response_data["start_date"] = survey.start_date
        response_data["end_date"] = survey.end_date

        if include_per_question_stats:
            try:
                from_ts = datetime.fromisoformat(date_from) if date_from else None
                to_ts = datetime.fromisoformat(date_to) if date_to else None
            except ValueError:
                from_ts = None
                to_ts = None
            per_question = fetch_per_question_stats(survey=survey, team=self.team, since=from_ts, until=to_ts)
            response_data["per_question_stats"] = [
                {
                    "question_id": q.question_id,
                    "question_index": q.question_index,
                    "question_text": q.question_text,
                    "question_type": q.question_type,
                    "response_count": q.response_count,
                    "distribution": q.distribution,
                    "average": q.average,
                }
                for q in per_question
            ]

        return Response(response_data)

    @action(
        methods=["POST"],
        detail=True,
        url_path="responses/(?P<response_uuid>[^/]+)/archive",
        required_scopes=["survey:write"],
    )
    def archive_response(self, request: request.Request, response_uuid: str, **kwargs) -> Response:
        """Archive a single survey response."""
        survey = self.get_object()

        try:
            UUID(response_uuid)
        except ValueError:
            return Response({"detail": "Invalid UUID format"}, status=400)

        archive, created = SurveyResponseArchive.objects.get_or_create(
            team_id=self.team_id,
            survey=survey,
            response_uuid=response_uuid,
        )

        if created:
            log_activity(
                organization_id=self.organization.id,
                team_id=self.team_id,
                user=cast(User, request.user),
                was_impersonated=is_impersonated(request),
                item_id=survey.id,
                scope="Survey",
                activity="response_archived",
                detail=Detail(name=f"Response {response_uuid}"),
            )

        return Response(status.HTTP_200_OK)

    @action(
        methods=["POST"],
        detail=True,
        url_path="responses/(?P<response_uuid>[^/]+)/unarchive",
        required_scopes=["survey:write"],
    )
    def unarchive_response(self, request: request.Request, response_uuid: str, **kwargs) -> Response:
        """Unarchive a single survey response."""
        survey = self.get_object()

        try:
            UUID(response_uuid)
        except ValueError:
            return Response({"detail": "Invalid UUID format"}, status=400)

        deleted_count, _ = SurveyResponseArchive.objects.filter(
            team_id=self.team_id, survey=survey, response_uuid=response_uuid
        ).delete()

        if deleted_count > 0:
            log_activity(
                organization_id=self.organization.id,
                team_id=self.team_id,
                user=cast(User, request.user),
                was_impersonated=is_impersonated(request),
                item_id=survey.id,
                scope="Survey",
                activity="response_unarchived",
                detail=Detail(name=f"Response {response_uuid}"),
            )

        return Response(status.HTTP_200_OK)

    @action(methods=["GET"], detail=True, url_path="archived-response-uuids", required_scopes=["survey:read"])
    def archived_response_uuids(self, request: request.Request, **kwargs) -> Response:
        """
        Get list of archived response UUIDs for InsightsQL filtering.

        Returns list of UUIDs that the frontend can use to filter out archived responses
        in InsightsQL queries.
        """
        survey = self.get_object()
        uuids = get_archived_response_uuids(str(survey.id), self.team_id)
        return Response(list(uuids))

    @extend_schema(
        operation_id="surveys_responses_list",
        description=(
            "List survey responses for a specific survey, with question text resolved server-side so callers do not "
            "have to map opaque `$survey_response_<id>` keys. Each row carries `distinct_id`, `session_id`, "
            "`submitted_at`, and an `extra` block (device, browser, OS, geoip, current_url, iteration) so agents can "
            "cross-pivot to recordings, persons, or paths in a single follow-up call. For person properties at "
            "event time, follow up with `persons-get` using the returned `distinct_id` — keeps scopes scoped. "
            "Use `question_id` + `score_lte` to fetch NPS detractors and similar score-filtered cohorts."
        ),
        parameters=[SurveyResponsesQuerySerializer],
        responses={200: SurveyResponsesListSerializer},
    )
    @action(methods=["GET"], detail=True, url_path="responses", required_scopes=["survey:read", "query:read"])
    def survey_responses_list(self, request: request.Request, **kwargs) -> Response:
        survey = self.get_object()

        params = SurveyResponsesQuerySerializer(data=request.query_params)
        params.is_valid(raise_exception=True)
        cleaned = params.validated_data

        # score_lte / score_gte only make sense for a specific rating question — reject the
        # combination at the API boundary rather than letting the InsightsQL helper raise later.
        if (cleaned.get("score_lte") is not None or cleaned.get("score_gte") is not None) and not cleaned.get(
            "question_id"
        ):
            raise exceptions.ValidationError("score_lte and score_gte require question_id to be specified")

        exclude_uuids: set[str] | None = None
        if cleaned.get("exclude_archived"):
            exclude_uuids = get_archived_response_uuids(str(survey.id), self.team_id)

        rows, has_more = fetch_response_rows(
            survey=survey,
            team=self.team,
            since=cleaned.get("since"),
            until=cleaned.get("until"),
            question_id=cleaned.get("question_id"),
            score_lte=cleaned.get("score_lte"),
            score_gte=cleaned.get("score_gte"),
            limit=cleaned.get("limit", 100),
            offset=cleaned.get("offset", 0),
            exclude_uuids=exclude_uuids,
        )

        tag_queries(product=ProductKey.SURVEYS, feature=Feature.QUERY)

        serialized = SurveyResponsesListSerializer(
            instance={
                "results": [
                    {
                        "uuid": row.uuid,
                        "distinct_id": row.distinct_id,
                        "session_id": row.session_id,
                        "submitted_at": row.submitted_at,
                        "answers": [
                            {
                                "question_id": a.question_id,
                                "question_index": a.question_index,
                                "question_text": a.question_text,
                                "question_type": a.question_type,
                                "answer": a.answer,
                            }
                            for a in row.answers
                        ],
                        "extra": row.extra,
                    }
                    for row in rows
                ],
                "has_more": has_more,
                "limit": cleaned.get("limit", 100),
                "offset": cleaned.get("offset", 0),
            }
        )
        return Response(serialized.data)

    @extend_schema(
        operation_id="surveys_question_labels",
        description=(
            "Return a slim list of question labels for the team's surveys. Used by the frontend to resolve "
            "`$survey_response_<question_id>` property keys into human-readable question text without "
            "loading the full survey payload."
        ),
        responses={
            200: inline_serializer(
                name="SurveyQuestionLabelsResponse",
                fields={
                    "labels": serializers.ListField(
                        child=inline_serializer(
                            name="SurveyQuestionLabel",
                            fields={
                                "question_id": serializers.CharField(help_text="UUID assigned to the survey question."),
                                "question_text": serializers.CharField(
                                    help_text="Untranslated question text as configured by the survey author.",
                                    allow_blank=True,
                                ),
                                "question_index": serializers.IntegerField(
                                    help_text="Zero-based index of the question within the survey."
                                ),
                                "survey_id": serializers.CharField(
                                    help_text="UUID of the survey this question belongs to."
                                ),
                                "survey_name": serializers.CharField(help_text="Display name of the survey."),
                            },
                        ),
                        help_text="One entry per question that has an ID assigned, across all the team's surveys.",
                    ),
                },
            ),
        },
    )
    @action(methods=["GET"], detail=False, url_path="question_labels", required_scopes=["survey:read"])
    def question_labels(self, request: request.Request, **kwargs) -> Response:
        # Custom (non-`list`) actions skip the routing layer's automatic access-level filter,
        # so apply it explicitly here — otherwise a user with access to only one survey would
        # receive labels for every survey in the team.
        queryset = self.user_access_control.filter_queryset_by_access_level(self.get_queryset())
        # The viewset's class-level queryset pre-joins `linked_flag`, `linked_insight`,
        # `targeting_flag`, `internal_targeting_flag` via `select_related`. `.only(...)` on
        # those deferred FK columns raises `FieldError: cannot be both deferred and traversed
        # using select_related`. Reset the select_related list before slimming the projection.
        queryset = queryset.select_related(None).only("id", "name", "questions")
        labels: list[dict[str, Any]] = []
        for survey in queryset.iterator(chunk_size=200):
            questions = survey.questions or []
            if not isinstance(questions, list):
                continue
            for index, question in enumerate(questions):
                if not isinstance(question, dict):
                    continue
                question_id = question.get("id")
                if not question_id:
                    continue
                labels.append(
                    {
                        "question_id": question_id,
                        "question_text": question.get("question") or "",
                        "question_index": index,
                        "survey_id": str(survey.id),
                        "survey_name": survey.name,
                    }
                )
        return Response({"labels": labels})

    @extend_schema(
        parameters=[
            OpenApiParameter(
                "date_from",
                OpenApiTypes.DATETIME,
                required=False,
                description="Optional ISO timestamp for start date (e.g. 2024-01-01T00:00:00Z)",
            ),
            OpenApiParameter(
                "date_to",
                OpenApiTypes.DATETIME,
                required=False,
                description="Optional ISO timestamp for end date (e.g. 2024-01-31T23:59:59Z)",
            ),
        ],
        responses={
            200: inline_serializer(
                name="SurveyGlobalStatsResponse",
                fields={
                    "stats": serializers.DictField(
                        help_text="Event counts keyed by event name (survey shown, survey dismissed, survey sent)."
                    ),
                    "rates": serializers.DictField(help_text="Calculated response and dismissal rates."),
                },
            )
        },
    )
    @extend_schema(operation_id="surveys_global_stats_retrieve")
    @action(methods=["GET"], detail=False, url_path="stats", required_scopes=["survey:read"])
    def global_stats(self, request: request.Request, **kwargs) -> Response:
        """Get aggregated response statistics across all surveys.

        Args:
            date_from: Optional ISO timestamp for start date (e.g. 2024-01-01T00:00:00Z)
            date_to: Optional ISO timestamp for end date (e.g. 2024-01-31T23:59:59Z)

        Returns:
            Aggregated statistics across all surveys including total counts and rates
        """
        date_from = request.query_params.get("date_from", None)
        date_to = request.query_params.get("date_to", None)

        response_data = self._get_survey_stats(date_from, date_to)
        return Response(response_data)

    @extend_schema(operation_id="surveys_all_activity_retrieve")
    @action(methods=["GET"], url_path="activity", detail=False, required_scopes=["activity_log:read"])
    def all_activity(self, request: request.Request, **kwargs):
        limit = int(request.query_params.get("limit", "10"))
        page = int(request.query_params.get("page", "1"))

        activity_page = load_activity(scope="Survey", team_id=self.team_id, limit=limit, page=page)

        return activity_page_response(activity_page, limit, page, request)

    @action(methods=["GET"], detail=True, required_scopes=["activity_log:read"])
    def activity(self, request: request.Request, **kwargs):
        limit = int(request.query_params.get("limit", "10"))
        page = int(request.query_params.get("page", "1"))

        item = self.get_object()

        activity_page = load_activity(
            scope="Survey",
            team_id=self.team_id,
            item_ids=[str(item.id)],
            limit=limit,
            page=page,
        )
        return activity_page_response(activity_page, limit, page, request)

    @extend_schema(
        description=(
            "Summarize survey responses. When `question_index` or `question_id` is provided, returns a per-question "
            "theme summary using cached `survey.question_summaries` when fresh. When neither is provided, returns "
            "the survey-wide headline summary (delegates to summary_headline). Pass `force_refresh=true` in the body "
            "to bypass caches."
        ),
        # request= is critical here — without it drf-spectacular falls back to the default ModelViewSet serializer
        # (SurveySerializerCreateUpdateOnly) and generates a Zod schema demanding name/type fields that have nothing
        # to do with summarization.
        request=SurveySummarizeRequestSerializer,
        parameters=[
            OpenApiParameter(
                "question_index",
                OpenApiTypes.INT,
                required=False,
                description="Zero-based question index. Omit to get the survey-wide headline instead.",
            ),
            OpenApiParameter(
                "question_id",
                OpenApiTypes.STR,
                required=False,
                description="Question UUID. Preferred over question_index — stable across question edits.",
            ),
        ],
    )
    @action(methods=["POST"], detail=True, required_scopes=["survey:read"])
    def summarize_responses(self, request: request.Request, **kwargs):
        if not request.user.is_authenticated:
            raise exceptions.NotAuthenticated()

        user = cast(User, request.user)

        survey_id = kwargs["pk"]

        if not Survey.objects.filter(id=survey_id, team__project_id=self.project_id).exists():
            return Response(status=status.HTTP_404_NOT_FOUND)

        survey = self.get_object()

        try:
            question_index_param = request.query_params.get("question_index", None)
            question_index = int(question_index_param) if question_index_param else None
        except (ValueError, TypeError):
            question_index = None

        question_id = request.query_params.get("question_id", None)

        if question_index is None and question_id is None:
            # No question specified — dispatch to the survey-wide headline summarizer.
            # Keeps a single MCP tool surface for both per-question and whole-survey summarization.
            return self.summary_headline(request, **kwargs)

        # Check for force_refresh flag in request body
        force_refresh = request.data.get("force_refresh", False)

        # Check for cached summary in question_summaries field
        if not force_refresh and question_id and survey.question_summaries:
            cached_summary = survey.question_summaries.get(question_id)
            if cached_summary:
                return Response(
                    {
                        "content": cached_summary.get("summary", ""),
                        "response_count": cached_summary.get("responseCount", 0),
                        "generated_at": cached_summary.get("generatedAt"),
                        "trace_id": cached_summary.get("traceId"),
                        "cached": True,
                    }
                )

        # Short-term Redis cache for rapid repeated requests
        cache_key = f"summarize_survey_responses_{self.team.pk}_{self.kwargs['pk']}_{question_id or question_index}"
        cached_response = cache.get(cache_key)
        if cached_response is not None and not force_refresh:
            return Response(cached_response)

        environment_is_allowed = settings.DEBUG or is_cloud()
        has_gemini_api_key = bool(settings.GEMINI_API_KEY)
        if not environment_is_allowed or not has_gemini_api_key:
            raise exceptions.ValidationError("survey summary is only supported in Insights Cloud")

        # Same AI-processing gate that summary_headline uses — applies to per-question summaries too.
        if not self.team.organization.is_ai_data_processing_approved:
            return Response(
                {"error": "AI data processing must be approved to generate summaries"},
                status=status.HTTP_403_FORBIDDEN,
            )

        end_date: datetime = (survey.end_date or datetime.now()).replace(
            hour=0, minute=0, second=0, microsecond=0
        ) + timedelta(days=1)

        # Extract the question text and choices from the survey
        question_text = None
        question_dict = None
        if survey.questions and question_id:
            # Find the question with the matching ID
            for idx, question in enumerate(survey.questions):
                if question.get("id", None) == question_id:
                    question_text = question.get("question")
                    question_dict = question
                    # Backfill the index so the index-based response key fallback works.
                    # Without this, fetch_responses passes question_index=None into the
                    # getSurveyResponse() InsightsQL function, which requires an integer first
                    # argument and raises QueryError -> 500.
                    if question_index is None:
                        question_index = idx
                    break
        elif survey.questions and question_index is not None:
            # Fallback to question index if question_id is not provided
            if 0 <= question_index < len(survey.questions):
                question_dict = survey.questions[question_index]
                question_text = question_dict.get("question")

        if question_text is None:
            raise exceptions.ValidationError("the text of the question is required")

        # For choice questions, exclude predefined choices (base and translated) to only get
        # open-ended "Other" responses — otherwise a predefined answer given in a non-base
        # language leaks into the free-text set fed to the summarizer. Same translation-aware
        # matching as per_question_stats' distribution.
        exclude_values = None
        if question_dict is not None:
            exclude_values = list(build_choice_translation_map(question_dict).keys()) or None

        # Get archived response UUIDs to exclude
        archived_uuids = get_archived_response_uuids(survey_id, self.team.pk)

        # Fetch responses using the new module
        responses = fetch_responses(
            survey_id=survey_id,
            question_index=question_index,
            question_id=question_id,
            start_date=(survey.start_date or survey.created_at).replace(hour=0, minute=0, second=0, microsecond=0),
            end_date=end_date,
            team=self.team,
            exclude_values=exclude_values,
            exclude_uuids=archived_uuids,
        )
        response_count = len(responses)

        if not responses:
            return Response(
                {
                    "content": "No responses to analyze.",
                    "response_count": 0,
                    "generated_at": datetime.now(UTC).isoformat(),
                    "cached": False,
                }
            )

        # Generate summary using Gemini
        try:
            result = summarize_responses(
                question_text,
                responses,
                distinct_id=str(user.distinct_id),
                survey_id=str(survey_id),
                question_id=question_id,
                team_id=self.team.pk,
            )
            content = format_as_markdown(result.summary)
            trace_id = result.trace_id
        except exceptions.ValidationError:
            raise
        except exceptions.APIException:
            raise
        except Exception as e:
            logger.exception(
                "Failed to generate survey summary",
                survey_id=survey_id,
                question_id=question_id,
                error_type=type(e).__name__,
                error=str(e),
            )
            raise exceptions.APIException("Failed to generate summary. Please try again.")

        generated_at = datetime.now(UTC).isoformat()

        # Prepare response data
        response_data = {
            "content": content,
            "response_count": response_count,
            "generated_at": generated_at,
            "trace_id": trace_id,
            "cached": False,
        }

        # Save to question_summaries field for long-term caching
        if question_id:
            if survey.question_summaries is None:
                survey.question_summaries = {}
            survey.question_summaries[question_id] = {
                "summary": content,
                "responseCount": response_count,
                "generatedAt": generated_at,
                "traceId": trace_id,
            }
            survey.save(update_fields=["question_summaries"])

        # Short-term Redis cache
        cache.set(cache_key, response_data, timeout=30)

        hanzo_insights.capture(
            event="survey response summarized",
            distinct_id=str(user.distinct_id),
            properties={"survey_id": survey_id, "question_id": question_id, "response_count": response_count},
        )

        # let the browser cache for half the time we cache on the server
        return Response(response_data, headers={"Cache-Control": "max-age=15"})

    @extend_schema(
        request=GenerateSurveyTranslationsRequestSerializer,
        responses=GenerateSurveyTranslationsResponseSerializer,
    )
    @action(methods=["POST"], detail=True, url_path="generate_translations", required_scopes=["survey:write"])
    def generate_translations(self, request: request.Request, **kwargs):
        if not request.user.is_authenticated:
            raise exceptions.NotAuthenticated()

        if not (settings.DEBUG or is_cloud()):
            raise exceptions.ValidationError(
                "survey translation generation is only supported in Insights Cloud or DEBUG mode"
            )

        if not settings.GEMINI_API_KEY:
            raise exceptions.ValidationError("GEMINI_API_KEY must be configured to generate translations")

        if not self.team.organization.is_ai_data_processing_approved:
            return Response(
                {"error": "AI data processing must be approved to generate translations"},
                status=status.HTTP_403_FORBIDDEN,
            )

        saved_survey = self.get_object()

        serializer = GenerateSurveyTranslationsRequestSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data
        saved_survey = self.get_object()
        survey = data.get("survey")
        if survey is None:
            survey = {
                "name": saved_survey.name,
                "description": saved_survey.description,
                "type": saved_survey.type,
                "appearance": saved_survey.appearance or {},
                "questions": saved_survey.questions or [],
                "translations": saved_survey.translations or {},
            }
        user = cast(User, request.user)

        source_language = (
            data.get("source_language")
            or getattr(saved_survey, "base_language", DEFAULT_BASE_LANGUAGE)
            or DEFAULT_BASE_LANGUAGE
        )

        translations, questions, generated_field_paths, trace_id = generate_survey_translation(
            survey=survey,
            target_language=data["target_language"],
            source_language=source_language,
            overwrite=data["overwrite"],
            distinct_id=str(user.distinct_id),
            team_id=self.team.pk,
        )

        hanzo_insights.capture(
            event="survey translations generated",
            distinct_id=str(user.distinct_id),
            properties={
                "survey_id": kwargs["pk"],
                "target_language": data["target_language"],
                "field_count": len(generated_field_paths),
            },
        )
        return Response(
            {
                "translations": translations,
                "questions": questions,
                "generated_field_paths": generated_field_paths,
                "trace_id": trace_id,
            }
        )

    @action(methods=["POST"], detail=True, required_scopes=["survey:write"])
    def duplicate_to_projects(self, request: request.Request, **kwargs):
        """Duplicate a survey to multiple projects in a single transaction.

        Accepts a list of target team IDs and creates a copy of the survey in each project.
        Uses an all-or-nothing approach - if any duplication fails, all changes are rolled back.
        """
        if not request.user.is_authenticated:
            raise exceptions.NotAuthenticated()

        user = cast(User, request.user)
        survey_id = kwargs["pk"]

        if not Survey.objects.filter(id=survey_id, team__project_id=self.project_id).exists():
            return Response(status=status.HTTP_404_NOT_FOUND)

        source_survey = self.get_object()
        target_team_ids = request.data.get("target_team_ids", [])

        if not target_team_ids or not isinstance(target_team_ids, list):
            raise exceptions.ValidationError("target_team_ids must be a non-empty list of team IDs")

        user_organization = user.organization
        if not user_organization:
            raise exceptions.ValidationError("User must belong to an organization")

        target_teams = Team.objects.filter(id__in=target_team_ids, organization_id=user_organization.id)

        if len(target_teams) != len(target_team_ids):
            raise exceptions.ValidationError("One or more target teams not found or you don't have access to them")

        duplicate_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        created_surveys = []

        try:
            with transaction.atomic():
                # Validate all survey data first
                surveys_to_create = []
                for team in target_teams:
                    cleaned_conditions = None
                    if source_survey.conditions:
                        cleaned_conditions = dict(source_survey.conditions)
                        # Remove project-specific fields: flags, actions, and events don't transfer across projects
                        cleaned_conditions.pop("linkedFlagVariant", None)
                        cleaned_conditions.pop("actions", None)
                        cleaned_conditions.pop("events", None)

                    survey_data = {
                        "name": f"{source_survey.name} (duplicated at {duplicate_timestamp})",
                        "description": source_survey.description,
                        "type": source_survey.type,
                        "questions": [
                            {k: v for k, v in q.items() if k != "id"} for q in (source_survey.questions or [])
                        ],
                        "appearance": source_survey.appearance,
                        "conditions": cleaned_conditions,
                        "archived": False,
                        "start_date": None,
                        "end_date": None,
                        "responses_limit": source_survey.responses_limit,
                        "iteration_count": source_survey.iteration_count,
                        "iteration_frequency_days": source_survey.iteration_frequency_days,
                        "schedule": source_survey.schedule,
                        "enable_partial_responses": source_survey.enable_partial_responses,
                    }
                    # NOTE: linked_flag_id, targeting_flag, and internal flags are intentionally omitted
                    # as they are project-specific and don't exist in other projects

                    serializer = SurveySerializerCreateUpdateOnly(
                        data=survey_data,
                        context={
                            "request": request,
                            "team_id": team.id,
                            "project_id": team.project_id,
                            "get_team": lambda t=team: t,
                            "get_organization": lambda: self.organization,
                        },
                    )

                    serializer.is_valid(raise_exception=True)

                    # Build Survey instance from validated data
                    new_survey = Survey(
                        team=team,
                        created_by=user,
                        **serializer.validated_data,
                    )
                    surveys_to_create.append(new_survey)

                for survey in surveys_to_create:
                    ensure_question_ids(survey)

                # Bulk create all surveys
                created_survey_objects = Survey.objects.bulk_create(surveys_to_create)

                # Prepare response data and activity logs
                for created_survey in created_survey_objects:
                    created_surveys.append(
                        {
                            "team_id": created_survey.team_id,
                            "survey_id": str(created_survey.id),
                            "name": created_survey.name,
                        }
                    )

                    log_activity(
                        organization_id=user_organization.id,
                        team_id=created_survey.team_id,
                        user=user,
                        was_impersonated=is_impersonated(request),
                        item_id=created_survey.id,
                        scope="Survey",
                        activity="created",
                        detail=Detail(
                            name=created_survey.name,
                            changes=[
                                Change(
                                    type="Survey",
                                    action="created",
                                    field="source_survey_id",
                                    after=str(survey_id),
                                )
                            ],
                        ),
                    )

        except exceptions.ValidationError as e:
            structlog.get_logger(__name__).error(
                "bulk_survey_duplication_validation_error",
                error=str(e),
                error_detail=e.detail if hasattr(e, "detail") else None,
                survey_id=survey_id,
                user_id=user.id,
                target_team_ids=target_team_ids,
            )
            raise
        except Exception as e:
            structlog.get_logger(__name__).error(
                "bulk_survey_duplication_error",
                error=str(e),
                error_type=type(e).__name__,
                survey_id=survey_id,
                user_id=user.id,
                target_team_ids=target_team_ids,
            )
            capture_exception(e)
            raise exceptions.ValidationError(
                {
                    "error": "Bulk duplication failed due to an unexpected error",
                }
            )

        hanzo_insights.capture(
            event="survey bulk duplicated",
            distinct_id=str(user.distinct_id),
            properties={
                "source_survey_id": str(survey_id),
                "target_count": len(created_surveys),
                "target_team_ids": target_team_ids,
            },
        )

        return Response(
            {
                "created_surveys": created_surveys,
                "count": len(created_surveys),
            },
            status=status.HTTP_201_CREATED,
        )


class SurveyAPIActionSerializer(serializers.ModelSerializer):
    steps = ActionStepJSONSerializer(many=True, required=False)

    class Meta:
        model = Action
        fields = [
            "id",
            "name",
            "steps",
        ]
        read_only_fields = fields


def _strip_invalid_translation_keys(translations: Any, normalized_base: str) -> dict[str, Any] | None:
    """Filter a translation map (survey-level or question-level) down to keys the SDK can match.

    Drops sentinels ('default', 'original', 'base'), non-BCP-47 shapes, and keys whose normalized
    form equals the survey's base language (since the SDK renders base text in that case).
    """
    if not isinstance(translations, dict):
        return None
    filtered: dict[str, Any] = {}
    for language, translation in translations.items():
        if not isinstance(language, str) or not isinstance(translation, dict):
            continue
        normalized = _normalize_language_code(language)
        if (
            normalized in REJECTED_TRANSLATION_KEYS
            or not BCP47_LANGUAGE_CODE_RE.match(normalized)
            or normalized == normalized_base
        ):
            continue
        filtered[normalized] = translation
    return filtered or None


def get_survey_api_translations(
    translations: Any, base_language: str = DEFAULT_BASE_LANGUAGE
) -> dict[str, dict[str, str]] | None:
    if not isinstance(translations, dict):
        return None

    normalized_base = _normalize_language_code(base_language) if isinstance(base_language, str) else ""
    safe_translations: dict[str, dict[str, str]] = {}
    for language, translation in translations.items():
        if not isinstance(language, str) or not isinstance(translation, dict):
            continue

        # Drop entries the SDK can't match: invalid codes, sentinels like "default", or the base language itself.
        normalized = _normalize_language_code(language)
        if (
            normalized in REJECTED_TRANSLATION_KEYS
            or not BCP47_LANGUAGE_CODE_RE.match(normalized)
            or normalized == normalized_base
        ):
            continue

        safe_translation = {
            field: value
            for field, value in translation.items()
            if field in SURVEY_API_TRANSLATION_FIELDS and isinstance(value, str)
        }
        if safe_translation:
            safe_translations[normalized] = safe_translation

    return safe_translations or None


class SurveyAPISerializer(serializers.ModelSerializer):
    """
    Serializer for the exposed /api/surveys endpoint, to be used in insights-js and for headless APIs.
    """

    linked_flag_key = serializers.CharField(source="linked_flag.key", read_only=True)
    targeting_flag_key = serializers.CharField(source="targeting_flag.key", read_only=True)
    internal_targeting_flag_key = serializers.CharField(source="internal_targeting_flag.key", read_only=True)
    conditions = serializers.SerializerMethodField(method_name="get_conditions")
    enable_partial_responses = serializers.BooleanField(read_only=True)
    questions = serializers.SerializerMethodField(method_name="get_questions")
    translations = serializers.SerializerMethodField(method_name="get_translations")

    class Meta:
        model = Survey
        fields = [
            "id",
            "name",
            # NB: The "description" field is serialized on Create/Update request, and used to be serialized on the next line,
            # But we had a user write in complaining that we were exposing the description in the API
            # (https://insightshelp.zendesk.com/agent/tickets/15210), which was a problem for them
            # since they were using it as a way to store sensitive information. Given that we don't ever use
            # that field to render the survey, we can safely remove it from the API response.
            "type",
            "linked_flag_key",
            "targeting_flag_key",
            "internal_targeting_flag_key",
            "questions",
            "conditions",
            "appearance",
            "start_date",
            "end_date",
            "current_iteration",
            "current_iteration_start_date",
            "schedule",
            "enable_partial_responses",
            "translations",
        ]
        read_only_fields = fields

    @extend_schema_field(serializers.DictField(allow_null=True))
    def get_conditions(self, survey: Survey):
        return get_survey_conditions_with_actions(survey, SurveyAPIActionSerializer)

    @extend_schema_field(
        serializers.DictField(child=serializers.DictField(child=serializers.CharField()), allow_null=True)
    )
    def get_translations(self, survey: Survey) -> dict[str, dict[str, str]] | None:
        return get_survey_api_translations(
            survey.translations, getattr(survey, "base_language", DEFAULT_BASE_LANGUAGE) or DEFAULT_BASE_LANGUAGE
        )

    @extend_schema_field(serializers.ListField(child=serializers.DictField(), allow_null=True))
    def get_questions(self, survey: Survey) -> list[dict[str, Any]] | None:
        """Return only question fields used by SDKs, with translation keys normalized."""
        questions = survey.questions
        if not isinstance(questions, list):
            return questions
        normalized_base = _normalize_language_code(
            getattr(survey, "base_language", DEFAULT_BASE_LANGUAGE) or DEFAULT_BASE_LANGUAGE
        )
        cleaned: list[dict[str, Any]] = []
        for question in questions:
            if not isinstance(question, dict):
                cleaned.append(question)
                continue
            next_question = dict(question)
            next_question.pop("isNpsQuestion", None)
            inline_translations = question.get("translations")
            if isinstance(inline_translations, dict):
                filtered = _strip_invalid_translation_keys(inline_translations, normalized_base)
                if filtered:
                    next_question["translations"] = filtered
                else:
                    next_question.pop("translations", None)
            cleaned.append(next_question)
        return cleaned

    def to_representation(self, instance: Survey) -> dict[str, Any]:
        data = super().to_representation(instance)
        if data.get("translations") is None:
            data.pop("translations", None)
        return data


def get_surveys_opt_in(team: Team) -> bool:
    # return False if the team has not set a value for surveys_opt_in
    if team.surveys_opt_in is None:
        return False
    return team.surveys_opt_in


def get_surveys_count(team: Team) -> int:
    return (
        Survey.objects.db_manager(READ_DB_FOR_SURVEYS)
        .filter(team__project_id=team.project_id)
        .exclude(archived=True)
        .count()
    )


def get_surveys_response(team: Team) -> dict[str, Any]:
    # Every SDK discards surveys that aren't running before evaluating eligibility, so drafts and
    # stopped surveys are payload every client downloads and throws away. The nullness check mirrors
    # the SDKs' own `isSurveyRunning`; keeping it time-independent matters because this response is
    # cached and only rebuilt when a survey is saved, so a comparison against `now` would go stale.
    surveys = SurveyAPISerializer(
        Survey.objects.db_manager(READ_DB_FOR_SURVEYS)
        .filter(team__project_id=team.project_id)
        .exclude(archived=True)
        .filter(start_date__isnull=False, end_date__isnull=True)
        .select_related("linked_flag", "targeting_flag", "internal_targeting_flag")
        .prefetch_related("actions"),
        many=True,
    ).data

    return {"surveys": surveys}


def _survey_page_site_url() -> str:
    """
    Origin for the hosted survey page's static assets. Assets must load from the app
    origin even when the page is served through a reverse-proxy domain, which doesn't
    serve Django staticfiles. A localhost SITE_URL (the unset default) would emit asset
    URLs no external browser can reach, so fall back to host-relative paths — those at
    least keep direct app-origin access working.
    """
    hostname = urlparse(settings.SITE_URL).hostname
    if hostname in (None, "localhost", "127.0.0.1", "::1"):
        return ""
    return settings.SITE_URL


def _survey_error_response(
    request: HttpRequest,
    *,
    error_title: str,
    error_message: str,
    status: int,
    appearance: dict[str, Any] | None = None,
) -> HttpResponse:
    context: dict[str, Any] = {
        "error_title": error_title,
        "error_message": error_message,
        "site_url": _survey_page_site_url(),
    }
    if appearance is not None:
        context["appearance"] = appearance
    return render(request, "surveys/error.html", context, status=status)


@csrf_exempt
def public_survey_page(request, survey_id: str):
    """
    Server-side rendered public survey page with security and performance optimizations
    """
    if request.method == "OPTIONS":
        return cors_response(request, HttpResponse(""))

    # Input validation
    if not UUIDT.is_valid_uuid(survey_id):
        logger.warning("survey_page_invalid_id", survey_id=survey_id)
        return _survey_error_response(
            request,
            error_title="Invalid request",
            error_message="The requested survey is not available.",
            status=400,
        )

    # Database query with minimal fields and timeout protection
    try:
        # nosemgrep: idor-lookup-without-team (public survey page, intentionally unauthenticated)
        survey = Survey.objects.select_related("team").get(id=survey_id)
    except Survey.DoesNotExist:
        logger.info("survey_page_not_found", survey_id=survey_id)
        # Use generic error message to prevent survey ID enumeration
        return _survey_error_response(
            request,
            error_title="Survey not available",
            error_message="The requested survey is not available.",
            status=404,
        )
    except Exception as e:
        logger.exception("survey_page_db_error", error=str(e), survey_id=survey_id)
        capture_exception(e)
        return _survey_error_response(
            request,
            error_title="Service unavailable",
            error_message="The service is temporarily unavailable. Please try again later.",
            status=503,
        )

    survey_is_running = (
        survey.start_date is not None and survey.start_date <= datetime.now(UTC) and survey.end_date is None
    )

    # Check survey availability (combine checks for consistent error message)
    if survey.archived or survey.type != Survey.SurveyType.EXTERNAL_SURVEY or not survey_is_running:
        logger.info(
            "survey_page_access_denied",
            survey_id=survey_id,
            archived=survey.archived,
            survey_type=survey.type,
        )
        # Pass appearance so the error page still shows the customer's brand.
        # Use 404 instead of 403 to prevent information leakage.
        return _survey_error_response(
            request,
            error_title="Feels quiet in here",
            error_message="This survey isn't taking responses right now. It might be closed, expired, or not live yet.",
            status=404,
            appearance=survey.appearance or {},
        )

    # Build project config
    project_config = {
        "api_host": request.build_absolute_uri("/").rstrip("/"),
        "token": survey.team.api_token,
    }

    if hasattr(survey.team, "ui_host") and survey.team.ui_host:
        project_config["ui_host"] = survey.team.ui_host

    serializer = SurveyAPISerializer(survey)
    survey_data = serializer.data
    context = {
        "name": survey.name,
        "survey_id": survey_id,
        "survey_data": survey_data,
        "project_config": project_config,
        "site_url": _survey_page_site_url(),
        "display_language": get_hosted_survey_display_language(request),
        "debug": settings.DEBUG,
        "embed_mode": request.GET.get("embed") == "true",
    }

    logger.info("survey_page_rendered", survey_id=survey_id, team_id=survey.team.id)

    response = render(request, "surveys/public_survey.html", context)

    if survey.enable_iframe_embedding:
        response.xframe_options_exempt = True
    else:
        response["X-Frame-Options"] = "DENY"

    # Cache headers
    response["Cache-Control"] = f"public, max-age={CACHE_TIMEOUT_SECONDS}"
    response["Vary"] = "Accept-Encoding"  # Enable compression caching

    return response


@contextmanager
def create_flag_with_survey_errors():
    # context manager to raise error with a different message when flag creation fails
    try:
        yield
    except serializers.ValidationError as e:
        # get the full details of the error to figure out if it's a behavioural cohort error
        error_details = e.get_full_details()
        raw_filters = error_details.get("filters", [{}]) if isinstance(error_details, dict) else [{}]
        filters = raw_filters if isinstance(raw_filters, list) else [raw_filters]
        matching_errors = [
            detail
            for detail in filters
            if isinstance(detail, dict) and detail.get("code") == BEHAVIOURAL_COHORT_FOUND_ERROR_CODE
        ]
        if matching_errors:
            original_detail = str(matching_errors[0].get("message"))
            raise serializers.ValidationError(
                detail=original_detail.replace("feature flags", "surveys"),
                code=BEHAVIOURAL_COHORT_FOUND_ERROR_CODE,
            )
        raise


def nh3_clean_with_allow_list(to_clean: str):
    return nh3.clean(
        to_clean,
        link_rel="noopener",
        tags={
            "a",
            "abbr",
            "acronym",
            "area",
            "article",
            "aside",
            "b",
            "bdi",
            "bdo",
            "blockquote",
            "br",
            "caption",
            "center",
            "cite",
            "code",
            "col",
            "colgroup",
            "data",
            "dd",
            "del",
            "details",
            "dfn",
            "div",
            "dl",
            "dt",
            "em",
            "figcaption",
            "figure",
            "footer",
            "h1",
            "h2",
            "h3",
            "h4",
            "h5",
            "h6",
            "header",
            "hgroup",
            "hr",
            "i",
            "img",
            "ins",
            "kbd",
            "li",
            "map",
            "mark",
            "nav",
            "ol",
            "p",
            "pre",
            "q",
            "rp",
            "rt",
            "rtc",
            "ruby",
            "s",
            "samp",
            "small",
            "span",
            "strike",
            "strong",
            "sub",
            "summary",
            "sup",
            "table",
            "tbody",
            "td",
            "th",
            "thead",
            "time",
            "tr",
            "tt",
            "u",
            "ul",
            "var",
            "wbr",
        },
        attributes={
            "*": {"style", "lang", "title", "width", "height"},
            # below are mostly defaults to ammonia, but we need to add them explicitly
            # because this python binding doesn't allow additive allowing
            "a": {"href", "hreflang", "target"},
            "bdo": {"dir"},
            "blockquote": {"cite"},
            "col": {"align", "char", "charoff", "span"},
            "colgroup": {"align", "char", "charoff", "span"},
            "del": {"cite", "datetime"},
            "hr": {"align", "size", "width"},
            "img": {"align", "alt", "height", "src", "width"},
            "ins": {"cite", "datetime"},
            "ol": {"start", "type"},
            "q": {"cite"},
            "table": {
                "align",
                "bgcolor",
                "border",
                "cellpadding",
                "cellspacing",
                "frame",
                "rules",
                "summary",
                "width",
            },
            "tbody": {"align", "char", "charoff", "valign"},
            "td": {
                "abbr",
                "align",
                "axis",
                "bgcolor",
                "char",
                "charoff",
                "colspan",
                "headers",
                "height",
                "nowrap",
                "rowspan",
                "scope",
                "valign",
                "width",
            },
            "tfoot": {"align", "char", "charoff", "valign"},
            "th": {
                "abbr",
                "align",
                "axis",
                "bgcolor",
                "char",
                "charoff",
                "colspan",
                "headers",
                "height",
                "nowrap",
                "rowspan",
                "scope",
                "valign",
                "width",
            },
            "thead": {"align", "char", "charoff", "valign"},
            "tr": {"align", "bgcolor", "char", "charoff", "valign"},
        },
    )