insights/insights/models/team/team.py

from datetime import timedelta
from decimal import Decimal
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Optional, cast
from zoneinfo import ZoneInfo

from django.conf import settings
from django.contrib.postgres.fields import ArrayField
from django.core.cache import cache
from django.core.validators import MaxValueValidator, MinLengthValidator, MinValueValidator
from django.db import connection, models, transaction
from django.db.models import QuerySet
from django.db.models.signals import post_delete, post_save

import pytz
import pydantic

from insights.cloud_utils import is_cloud
from insights.datastore.query_tagging import Feature, Product, tag_queries, tags_context
from insights.helpers.session_recording_playlist_templates import DEFAULT_PLAYLISTS
from insights.models.filters.filter import Filter
from insights.models.filters.mixins.utils import cached_property
from insights.models.filters.utils import GroupTypeIndex
from insights.models.instance_setting import get_instance_setting
from insights.models.organization import Organization, OrganizationMembership
from insights.models.signals import mutable_receiver, secret_api_token_rotated
from insights.models.utils import (
    KeyKind,
    UUIDTClassicModel,
    generate_random_token_project,
    generate_random_token_secret,
    mask_key_value,
    mint,
    sane_repr,
    validate_rate_limit,
)
from insights.ph_client import feature_enabled_or_false
from insights.rbac.decorators import field_access_control
from insights.session_recordings.models.session_recording_playlist import SessionRecordingPlaylist
from insights.settings.utils import get_list

# Relocated to the Django-free insights.week_start_day module so the InsightsQL engine can use it
# without booting Django; re-exported here for existing callers.
from insights.week_start_day import WeekStartDay  # noqa: F401

from products.customer_analytics.backend.facade.constants import DEFAULT_ACTIVITY_EVENT

from ...insightsql.modifiers import set_default_modifier_values
from ...schema_enums import CurrencyCode, PersonsOnEventsMode
from .extensions import get_or_create_team_extension
from .team_caching import get_team_in_cache, set_team_in_cache

if TYPE_CHECKING:
    from insights.schema import PathCleaningFilter

    from insights.models.user import User

TIMEZONES = [(tz, tz) for tz in pytz.all_timezones]

# TODO: DEPRECATED; delete when these attributes can be fully removed from `Team` model
DEPRECATED_ATTRS = (
    "plugins_opt_in",
    "opt_out_capture",
    "event_names",
    "event_names_with_usage",
    "event_properties",
    "event_properties_with_usage",
    "event_properties_numerical",
    # Replaced by the `revenue_analytics_config` extension; only the raw-SQL Postgres->Datastore
    # ETL still reads the column, and that bypasses this manager.
    "revenue_tracking_config",
)

# Django requires a list of tuples for choices
CURRENCY_CODE_CHOICES = [(code.value, code.value) for code in CurrencyCode]

# Intentionally asserting this here to guarantee we remember
# to rerun migrations when a new currency is added
# python manage.py makemigrations
assert len(CURRENCY_CODE_CHOICES) == 152

DEFAULT_CURRENCY = CurrencyCode.USD.value

ASYNC_USER_PRODUCT_LIST_SYNC_THRESHOLD = 100


# keep in sync with insights/frontend/src/scenes/project/Settings/ExtraTeamSettings.tsx
class AvailableExtraSettings:
    pass


class TeamManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().defer(*DEPRECATED_ATTRS)

    def create_with_data(self, *, initiating_user: Optional["User"], **kwargs) -> "Team":
        team = cast("Team", self.create(**kwargs))

        # Create internal/test users cohort and set test_account_filters to exclude it
        from products.cohorts.backend.models.cohort import get_or_create_internal_test_users_cohort

        initiating_user_email = initiating_user.email if initiating_user else None
        test_users_cohort = get_or_create_internal_test_users_cohort(team, initiating_user_email=initiating_user_email)
        team.test_account_filters = [
            {"key": "id", "type": "cohort", "value": test_users_cohort.pk, "operator": "not_in"}
        ]

        if kwargs.get("is_demo"):
            if initiating_user is None:
                raise ValueError("initiating_user must be provided when creating a demo team")
            team.save()
            # Defer Celery task if we're inside an atomic block (e.g., ensure_account_and_save)
            # to avoid sending task before transaction commits. In tests, Celery tasks run
            # synchronously so we can call directly (and on_commit doesn't run in TestCase).
            if connection.in_atomic_block and not settings.TEST:
                transaction.on_commit(lambda: team.kick_off_demo_data_generation(initiating_user))
            else:
                team.kick_off_demo_data_generation(initiating_user)
            return team  # Return quickly, as the demo data and setup will be created asynchronously

        # Get organization to apply defaults
        organization = kwargs.get("organization") or Organization.objects.get(id=kwargs.get("organization_id"))

        team.anonymize_ips = kwargs.get("anonymize_ips", organization.default_anonymize_ips)

        # Self-hosted deployments get 5-year session recording retention by default
        if not is_cloud():
            team.session_recording_retention_period = kwargs.get(
                "session_recording_retention_period",
                SessionRecordingRetentionPeriod.FIVE_YEARS,
            )

        if team.extra_settings is None:
            team.extra_settings = {}
        team.extra_settings.setdefault("recorder_script", "insights-recorder")

        # Create default dashboards (demo projects skip above)
        from insights.helpers.dashboard_templates import create_from_template  # noqa: PLC0415 — breaks team import cycle

        from products.dashboards.backend.models.dashboard import Dashboard  # noqa: PLC0415 — breaks team import cycle
        from products.dashboards.backend.models.dashboard_templates import (  # noqa: PLC0415 — breaks team import cycle
            DashboardTemplate,
        )

        template = DashboardTemplate.default_signup_template()
        dashboard = Dashboard.objects.db_manager(self.db).create(
            name="Your starter dashboard",
            pinned=True,
            team=team,
            description=template.dashboard_description or "",
        )
        create_from_template(dashboard, template)
        team.primary_dashboard = dashboard

        # Create default session recording playlists
        for playlist in DEFAULT_PLAYLISTS:
            SessionRecordingPlaylist.objects.create(
                team=team,
                name=str(playlist["name"]),
                filters=playlist["filters"],
                description=str(playlist.get("description", "")),
                type="filters",
            )
        team.save()

        # Add UserProductList for all users who have access to this new team
        # For large orgs, dispatch async to avoid request timeouts
        from insights.tasks.tasks import sync_user_product_lists_for_new_team

        user_count = OrganizationMembership.objects.filter(organization_id=team.organization_id).count()
        if user_count > ASYNC_USER_PRODUCT_LIST_SYNC_THRESHOLD:
            sync_user_product_lists_for_new_team.delay(team.id)
        else:
            sync_user_product_lists_for_new_team(team.id)

        return team

    def create(self, **kwargs):
        from ..project import Project

        with transaction.atomic(using=self.db):
            if "id" not in kwargs:
                kwargs["id"] = self.increment_id_sequence()
            if kwargs.get("project") is None and kwargs.get("project_id") is None:
                # If a parent project is not provided for this team, ensure there is one
                # This should be removed once environments are fully rolled out
                project_kwargs = {}
                if organization := kwargs.get("organization"):
                    project_kwargs["organization"] = organization
                elif organization_id := kwargs.get("organization_id"):
                    project_kwargs["organization_id"] = organization_id
                if name := kwargs.get("name"):
                    project_kwargs["name"] = name
                kwargs["project"] = Project.objects.db_manager(self.db).create(id=kwargs["id"], **project_kwargs)
            return super().create(**kwargs)

    def get_team_from_token(self, token: Optional[str]) -> Optional["Team"]:
        if not token:
            return None
        try:
            return Team.objects.get(api_token=token)
        except Team.DoesNotExist:
            return None

    def get_team_from_cache_or_token(self, token: Optional[str]) -> Optional["Team"]:
        if not token:
            return None
        try:
            team = get_team_in_cache(token)
            if team:
                return team

            team = Team.objects.get(api_token=token)
            set_team_in_cache(token, team)
            return team

        except Team.DoesNotExist:
            return None

    def get_team_from_cache_or_secret_api_token(self, secret_api_token: Optional[str]) -> Optional["Team"]:
        if not secret_api_token:
            return None
        try:
            team = get_team_in_cache(secret_api_token)
            if team:
                return team

            team = Team.objects.get(secret_api_token=secret_api_token)
            set_team_in_cache(secret_api_token, team)
            return team

        except Team.DoesNotExist:
            return None

    def increment_id_sequence(self) -> int:
        """Increment the `Team.id` field's sequence and return the latest value.

        Use only when actually neeeded to avoid wasting sequence values."""
        cursor = connection.cursor()
        cursor.execute("SELECT nextval('insights_team_id_seq')")
        result = cursor.fetchone()
        return result[0]


def get_default_data_attributes() -> list[str]:
    return ["data-attr"]


class CookielessServerHashMode(models.IntegerChoices):
    DISABLED = 0, "Disabled"
    STATELESS = 1, "Stateless"
    STATEFUL = 2, "Stateful"


class BusinessModel(models.TextChoices):
    B2B = "b2b", "B2B"
    B2C = "b2c", "B2C"
    OTHER = "other", "Other"


class SessionRecordingRetentionPeriod(models.TextChoices):
    THIRTY_DAYS = "30d", "30 Days"
    NINETY_DAYS = "90d", "90 Days"
    ONE_YEAR = "1y", "1 Year"
    FIVE_YEARS = "5y", "5 Years"


class Team(UUIDTClassicModel):
    """Team means "environment" (historically it meant "project", but now we have the parent Project model for that)."""

    class Meta:
        verbose_name = "environment (aka team)"
        verbose_name_plural = "environments (aka teams)"
        # Route forward-FK / related-object loads (e.g. `request.user.current_team`) through the
        # `objects` manager so they inherit its `.defer(*DEPRECATED_ATTRS)`. Without this, Django
        # falls back to a bare `Manager()` for related access and re-loads the fat deprecated
        # taxonomy columns on every such fetch — on the hot path that's every authenticated request.
        base_manager_name = "objects"
        constraints = [
            models.CheckConstraint(
                name="project_id_is_not_null",
                # We have this as a constraint rather than IS NOT NULL on the field, because setting IS NOT NULL cannot
                # be done without locking the table. By adding this constraint using Postgres's `NOT VALID` option
                # (via Django `AddConstraintNotValid()`) and subsequent `VALIDATE CONSTRAINT`, we avoid locking.
                condition=models.Q(project_id__isnull=False),
            ),
            models.CheckConstraint(
                name="llm_gateway_overspend_allowance_usd_in_range",
                # Mirrors OVERSPEND_ALLOWANCE_MIN/MAX_USD; null = unset stays valid. The Python
                # write surfaces validate too, but update/bulk_update/shell/raw bypass them.
                condition=models.Q(llm_gateway_overspend_allowance_usd__isnull=True)
                | models.Q(
                    llm_gateway_overspend_allowance_usd__gte=0,
                    llm_gateway_overspend_allowance_usd__lte=10000,
                ),
            ),
        ]

    objects: TeamManager = TeamManager()

    organization = models.ForeignKey(
        "insights.Organization",
        on_delete=models.CASCADE,
        related_name="teams",
        related_query_name="team",
    )
    # NOTE: The deletion is not cascade due to us wanting to first of all solve deletion properly before allowing cascading deletes
    parent_team = models.ForeignKey(
        "insights.Team",
        on_delete=models.SET_NULL,
        related_name="child_teams",
        related_query_name="child_team",
        null=True,
    )
    # NOTE: To be removed in favour of parent_team
    project = models.ForeignKey(
        "insights.Project",
        on_delete=models.CASCADE,
        related_name="teams",
        related_query_name="team",
    )
    api_token = models.CharField(
        max_length=200,
        unique=True,
        default=generate_random_token_project,
        validators=[MinLengthValidator(10, "Project's API token must be at least 10 characters long!")],
    )
    app_urls: ArrayField = field_access_control(
        ArrayField(models.CharField(max_length=200, null=True), default=list, blank=True), "web_analytics", "editor"
    )
    name = models.CharField(
        max_length=200,
        default="Default project",
        validators=[MinLengthValidator(1, "Project must have a name!")],
    )
    # DEPRECATED - not used anywhere and eligible for deletion
    slack_incoming_webhook = models.CharField(max_length=500, null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    anonymize_ips = field_access_control(models.BooleanField(default=False), "project", "admin")
    completed_snippet_onboarding = models.BooleanField(default=False)
    has_completed_onboarding_for = models.JSONField(null=True, blank=True)
    onboarding_tasks = models.JSONField(null=True, blank=True)
    ingested_event = models.BooleanField(default=False)
    ingested_production_event = models.BooleanField(default=False, db_default=False)
    ingested_production_event_last_checked_at = models.DateTimeField(null=True, blank=True)

    person_processing_opt_out = field_access_control(models.BooleanField(null=True, default=False), "project", "admin")
    secret_api_token = models.CharField(
        max_length=200,
        null=True,
        blank=True,
    )
    secret_api_token_backup = models.CharField(
        max_length=200,
        null=True,
        blank=True,
    )

    # Session recording
    session_recording_opt_in = field_access_control(models.BooleanField(default=False), "project", "admin")
    session_recording_sample_rate = field_access_control(
        models.DecimalField(
            # will store a decimal between 0 and 1 allowing up to 2 decimal places
            null=True,
            blank=True,
            max_digits=3,
            decimal_places=2,
            validators=[MinValueValidator(Decimal(0)), MaxValueValidator(Decimal(1))],
        ),
        "project",
        "admin",
    )
    session_recording_minimum_duration_milliseconds = field_access_control(
        models.IntegerField(
            null=True,
            blank=True,
            validators=[MinValueValidator(0), MaxValueValidator(30000)],
        ),
        "project",
        "admin",
    )
    session_recording_linked_flag = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")
    session_recording_network_payload_capture_config = field_access_control(
        models.JSONField(null=True, blank=True), "project", "admin"
    )
    session_recording_masking_config = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")
    session_recording_url_trigger_config = field_access_control(
        ArrayField(models.JSONField(null=True, blank=True), default=list, blank=True, null=True),
        "project",
        "admin",
    )
    session_recording_url_blocklist_config = field_access_control(
        ArrayField(models.JSONField(null=True, blank=True), default=list, blank=True, null=True),
        "project",
        "admin",
    )
    session_recording_event_trigger_config = field_access_control(
        ArrayField(models.TextField(null=True, blank=True), default=list, blank=True, null=True),
        "project",
        "admin",
    )
    session_recording_trigger_match_type_config = field_access_control(
        models.CharField(null=True, blank=True, max_length=24),
        "project",
        "admin",
    )
    session_recording_trigger_groups = field_access_control(
        models.JSONField(
            null=True,
            blank=True,
            help_text="V2 trigger groups configuration for session recording. If present, takes precedence over legacy trigger fields.",
        ),
        "project",
        "admin",
    )
    session_replay_config = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")
    session_recording_retention_period = field_access_control(
        models.CharField(
            max_length=3,
            choices=SessionRecordingRetentionPeriod,
            default=SessionRecordingRetentionPeriod.THIRTY_DAYS,
        ),
        "project",
        "admin",
    )

    # Events data retention in months — denormalized from the billing entitlement by sync_events_retention, read on
    # the InsightsQL hot path. Default 84 (7 years) grandfathers existing teams. db_default because insights_team is also
    # written by raw inserts in nodejs/rust and the test-schema builder.
    event_retention_months = field_access_control(
        models.PositiveSmallIntegerField(default=84, db_default=84),
        "project",
        "admin",
    )

    # Conversations
    conversations_enabled = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")
    conversations_settings = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")

    # Proactive tasks (#team-self-driving)
    proactive_tasks_enabled = models.BooleanField(null=True, blank=True)

    # Surveys
    survey_config = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")
    surveys_opt_in = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")

    # Product tours
    product_tours_opt_in = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")

    # Capture / Autocapture
    capture_console_log_opt_in = field_access_control(
        models.BooleanField(null=True, blank=True, default=True), "project", "admin"
    )
    capture_performance_opt_in = field_access_control(
        models.BooleanField(null=True, blank=True, default=True), "project", "admin"
    )
    capture_dead_clicks = field_access_control(
        models.BooleanField(null=True, blank=True, default=False), "project", "admin"
    )
    autocapture_opt_out = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")
    autocapture_web_vitals_opt_in = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")
    autocapture_web_vitals_allowed_metrics = field_access_control(
        models.JSONField(null=True, blank=True), "project", "admin"
    )
    autocapture_exceptions_opt_in = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")
    autocapture_exceptions_errors_to_ignore = field_access_control(
        models.JSONField(null=True, blank=True), "project", "admin"
    )

    # Logs
    logs_settings = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")

    # LLM gateway — per-team admission projection read by the llm-gateway Go
    # service via a purpose-built HyperCache blob (cache/team_tokens/<token>/
    # team_metadata/llm_gateway_policy.json). The gateway admits a team only
    # when llm_gateway_enabled_at is set and llm_gateway_revoked_at is null;
    # revoke wins over enable. Null enabled_at = not enrolled (default-deny);
    # null revoked_at = not revoked. Set by internal tooling/admin only.
    llm_gateway_enabled_at = models.DateTimeField(null=True, blank=True)
    llm_gateway_revoked_at = models.DateTimeField(null=True, blank=True)
    # Per-team USD floor the gateway lets a team dispatch past $0 down to. Projected into
    # gateway_credential.json as the `overspend_allowance_usd` wire key. Null = use the
    # gateway default; 0 = no allowance. Range [0, 10000] validated at the write surfaces.
    llm_gateway_overspend_allowance_usd = models.DecimalField(max_digits=20, decimal_places=6, null=True, blank=True)

    # Heatmaps
    heatmaps_opt_in = field_access_control(models.BooleanField(null=True, blank=True), "project", "admin")

    # Activity logs
    receive_org_level_activity_logs = field_access_control(
        models.BooleanField(null=True, blank=True, default=False), "project", "admin"
    )

    # Web analytics
    web_analytics_pre_aggregated_tables_enabled = field_access_control(
        models.BooleanField(default=False, null=True), "project", "admin"
    )
    web_analytics_pre_aggregated_tables_version = field_access_control(
        models.CharField(max_length=10, default="v2", null=True, choices=[("v1", "v1"), ("v2", "v2")]),
        "project",
        "admin",
    )

    # Feature flags
    flags_persistence_default = field_access_control(
        models.BooleanField(null=True, blank=True, default=False), "project", "admin"
    )
    feature_flag_confirmation_enabled = field_access_control(
        models.BooleanField(null=True, blank=True, default=False), "project", "admin"
    )
    feature_flag_confirmation_message = field_access_control(
        models.TextField(null=True, blank=True), "project", "admin"
    )
    # DEPRECATED: Use default_evaluation_contexts_enabled instead. Will be removed in a future migration after full rollout.
    default_evaluation_environments_enabled = models.BooleanField(null=True, blank=True, default=False)
    # DEPRECATED: Use require_evaluation_contexts instead. Will be removed in a future migration after full rollout.
    require_evaluation_environment_tags = models.BooleanField(null=True, blank=True, default=False)
    default_evaluation_contexts_enabled = field_access_control(
        models.BooleanField(
            null=True,
            blank=True,
            default=False,
            help_text="Whether to automatically apply default evaluation contexts to new feature flags",
        ),
        "project",
        "admin",
    )
    require_evaluation_contexts = field_access_control(
        models.BooleanField(
            null=True,
            blank=True,
            default=False,
            help_text="Whether to require at least one evaluation context tag when creating new feature flags",
        ),
        "project",
        "admin",
    )
    session_recording_version = field_access_control(
        models.CharField(null=True, blank=True, max_length=24), "project", "admin"
    )
    signup_token = models.CharField(max_length=200, null=True, blank=True)
    is_demo = field_access_control(models.BooleanField(default=False), "project", "admin")

    # DEPRECATED - do not use
    access_control = models.BooleanField(default=False)

    week_start_day = field_access_control(
        models.SmallIntegerField(null=True, blank=True, choices=WeekStartDay), "project", "admin"
    )
    # This is not a manual setting. It's updated automatically to reflect if the team uses site apps or not.
    inject_web_apps = models.BooleanField(null=True)

    test_account_filters = field_access_control(
        models.JSONField(
            default=list,
            help_text="""Filters used to identify internal/test users. Each entry is a property filter.

            Supported entry types and the exact shape each accepts:

            # Person property — match (or exclude) by a person property
            {"key": "email", "type": "person", "value": "@example.com", "operator": "icontains"}

            # Event property — match by an event property
            {"key": "$host", "type": "event", "value": "localhost", "operator": "icontains"}

            # Cohort membership — match (or exclude) members of a cohort.
            # Use operator "in" for inclusion and "not_in" for exclusion. Do NOT use a
            # `negation` field here — `negation` is specific to cohort *definitions*
            # (the inner sub-filters that build a cohort) and is rejected by the
            # property-filter schema.
            {"key": "id", "type": "cohort", "value": 8814, "operator": "not_in"}

            Common operators: "exact", "is_not", "icontains", "not_icontains", "regex",
            "not_regex", "gt", "lt", "gte", "lte", "is_set", "is_not_set", "in", "not_in".""",
        ),
        "project",
        "admin",
    )
    test_account_filters_default_checked = field_access_control(
        models.BooleanField(null=True, blank=True), "project", "admin"
    )

    path_cleaning_filters = field_access_control(
        models.JSONField(default=list, null=True, blank=True), "project", "admin"
    )

    timezone = field_access_control(
        models.CharField(max_length=240, choices=TIMEZONES, default="UTC"), "project", "admin"
    )
    data_attributes = field_access_control(
        models.JSONField(default=get_default_data_attributes, blank=True), "project", "admin"
    )
    person_display_name_properties: ArrayField = field_access_control(
        ArrayField(models.CharField(max_length=400), null=True, blank=True), "project", "admin"
    )
    live_events_columns: ArrayField = field_access_control(
        ArrayField(models.TextField(), null=True, blank=True), "project", "admin"
    )
    recording_domains: ArrayField = field_access_control(
        ArrayField(models.CharField(max_length=200, null=True), blank=True, null=True), "project", "admin"
    )
    human_friendly_comparison_periods = field_access_control(
        models.BooleanField(default=False, null=True, blank=True), "project", "admin"
    )
    cookieless_server_hash_mode = field_access_control(
        models.SmallIntegerField(
            default=CookielessServerHashMode.DISABLED,
            choices=CookielessServerHashMode,
            null=True,
        ),
        "project",
        "admin",
    )

    primary_dashboard = models.ForeignKey(
        "dashboards.Dashboard",
        on_delete=models.SET_NULL,
        null=True,
        related_name="primary_dashboard_teams",
        blank=True,
    )  # Dashboard shown on project homepage

    default_data_theme = field_access_control(models.IntegerField(null=True, blank=True), "project", "admin")

    # Generic field for storing any team-specific context
    # that likely doesn't deserve a dedicated column. Can be used for things like settings and overrides
    # during feature releases.
    extra_settings = models.JSONField(null=True, blank=True)

    # Environment-level default InsightsQL query modifiers
    modifiers = field_access_control(models.JSONField(null=True, blank=True), "project", "admin")

    # This is meant to be used as a stopgap until https://github.com/Insights/meta/pull/39 gets implemented
    # Switches _most_ queries to using distinct_id as aggregator instead of person_id
    @property
    def aggregate_users_by_distinct_id(self) -> bool:
        return str(self.pk) in get_list(get_instance_setting("AGGREGATE_BY_DISTINCT_IDS_TEAMS"))

    # This correlation_config is intended to be used initially for
    # `excluded_person_property_names` but will be used as a general config
    # repository for correlation related settings.
    # NOTE: we're not doing any schema checking here, just storing whatever is
    # thrown at us. Correlation code can handle schema related issues.
    correlation_config = field_access_control(models.JSONField(default=dict, null=True, blank=True), "project", "admin")

    # DEPRECATED, DISUSED: recordings on CH are cleared with Datastore's TTL
    session_recording_retention_period_days = models.IntegerField(null=True, default=None, blank=True)
    # DEPRECATED, DISUSED: plugins are enabled for everyone now
    plugins_opt_in = models.BooleanField(default=False)
    # DEPRECATED, DISUSED: replaced with env variable OPT_OUT_CAPTURE and User.anonymized_data
    opt_out_capture = models.BooleanField(default=False)
    # DEPRECATED: in favor of `EventDefinition` model
    event_names = models.JSONField(default=list, blank=True)
    event_names_with_usage = models.JSONField(default=list, blank=True)
    # DEPRECATED: in favor of `PropertyDefinition` model
    event_properties = models.JSONField(default=list, blank=True)
    event_properties_with_usage = models.JSONField(default=list, blank=True)
    event_properties_numerical = models.JSONField(default=list, blank=True)
    external_data_workspace_id = models.CharField(max_length=400, null=True, blank=True)
    external_data_workspace_last_synced_at = models.DateTimeField(null=True, blank=True)

    api_query_rate_limit = models.CharField(
        max_length=32,
        null=True,
        blank=True,
        help_text="Custom rate limit for InsightsQL API queries in #requests/{sec,min,hour,day}",
        validators=[validate_rate_limit],
    )

    # DEPRECATED: use `revenue_analytics_config` property instead
    revenue_tracking_config = models.JSONField(null=True, blank=True)

    # Duration for dropping events older than this threshold
    drop_events_older_than = models.DurationField(
        null=True,
        blank=True,
        validators=[MinValueValidator(timedelta(hours=1))],  # For safety minimum 1h
        help_text="Events older than this threshold will be dropped in ingestion. Empty means no timestamp restrictions.",
    )

    # Consolidated base currency for all analytics (revenue, marketing, etc.)
    base_currency = field_access_control(
        models.CharField(
            max_length=3,
            choices=CURRENCY_CODE_CHOICES,
            default=DEFAULT_CURRENCY,
            null=True,
            blank=True,
        ),
        "project",
        "admin",
    )

    experiment_recalculation_time = field_access_control(
        models.TimeField(
            null=True,
            blank=True,
            help_text="Time of day (UTC) when experiment metrics should be recalculated. If not set, uses the default recalculation time.",
        ),
        "project",
        "admin",
    )

    default_experiment_confidence_level = field_access_control(
        models.DecimalField(
            max_digits=3,
            decimal_places=2,
            null=True,
            blank=True,
            help_text="Default confidence level for new experiments in this environment. Valid values: 0.90, 0.95, 0.99.",
        ),
        "project",
        "admin",
    )

    default_experiment_stats_method = field_access_control(
        models.CharField(
            max_length=20,
            choices=Organization.DefaultExperimentStatsMethod,
            default=Organization.DefaultExperimentStatsMethod.BAYESIAN,
            help_text="Default statistical method for new experiments in this environment.",
            null=True,
            blank=True,
        ),
        "project",
        "admin",
    )

    business_model = field_access_control(
        models.CharField(
            max_length=10,
            choices=BusinessModel,
            null=True,
            blank=True,
            help_text="Whether this project serves B2B or B2C customers, used to optimize the UI layout.",
        ),
        "project",
        "admin",
    )

    # Before adding new fields here, read insights/models/team/README.md
    # Domain-specific config should use a Team Extension model instead.

    # TRANSITIONAL: These accessors exist for backward compat with existing
    # `team.<product>_config` call sites. New products should NOT add accessors
    # here — use get_or_create_team_extension() at call sites instead.

    @cached_property
    def revenue_analytics_config(self):
        from .team_revenue_analytics_config import TeamRevenueAnalyticsConfig

        return get_or_create_team_extension(self, TeamRevenueAnalyticsConfig)

    @cached_property
    def marketing_analytics_config(self):
        from .team_marketing_analytics_config import TeamMarketingAnalyticsConfig

        return get_or_create_team_extension(self, TeamMarketingAnalyticsConfig)

    @cached_property
    def customer_analytics_config(self):
        from products.customer_analytics.backend.facade.team_extension import TeamCustomerAnalyticsConfig

        return get_or_create_team_extension(
            self, TeamCustomerAnalyticsConfig, defaults={"activity_event": DEFAULT_ACTIVITY_EVENT}
        )

    @cached_property
    def workflows_config(self):
        from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig

        return get_or_create_team_extension(self, TeamWorkflowsConfig)

    @property
    def default_modifiers(self) -> dict:
        # Deferred: insights.schema (the pydantic models) stays off django.setup(),
        # where this model loads in every process.
        from insights.schema import InsightsQLQueryModifiers  # noqa: PLC0415

        modifiers = InsightsQLQueryModifiers()
        set_default_modifier_values(modifiers, self)
        return modifiers.model_dump()

    @property
    def person_on_events_mode(self) -> PersonsOnEventsMode:
        if self.modifiers and self.modifiers.get("personsOnEventsMode") is not None:
            # InsightsQL modifiers (which also act as the project-level setting) take precedence
            mode = PersonsOnEventsMode(self.modifiers["personsOnEventsMode"])
        else:
            # Otherwise use the flag-based default
            mode = self.person_on_events_mode_flag_based_default
        tag_queries(person_on_events_mode=mode)
        return mode

    @property
    def person_on_events_mode_flag_based_default(self) -> PersonsOnEventsMode:
        if self._person_on_events_person_id_override_properties_on_events:
            return PersonsOnEventsMode.PERSON_ID_OVERRIDE_PROPERTIES_ON_EVENTS

        if self._person_on_events_person_id_no_override_properties_on_events:
            return PersonsOnEventsMode.PERSON_ID_NO_OVERRIDE_PROPERTIES_ON_EVENTS

        return PersonsOnEventsMode.PERSON_ID_OVERRIDE_PROPERTIES_JOINED

    # KLUDGE: DO NOT REFERENCE IN THE BACKEND!
    # Keeping this property for now only to be used by the frontend in certain cases
    @property
    def person_on_events_querying_enabled(self) -> bool:
        return self.person_on_events_mode in (  # Whether person properties on events are in use by default
            PersonsOnEventsMode.PERSON_ID_OVERRIDE_PROPERTIES_ON_EVENTS,
            PersonsOnEventsMode.PERSON_ID_NO_OVERRIDE_PROPERTIES_ON_EVENTS,
        )

    @property
    def _person_on_events_person_id_no_override_properties_on_events(self) -> bool:
        person_on_events_override = getattr(settings, "PERSON_ON_EVENTS_OVERRIDE", None)
        if person_on_events_override is not None:
            return person_on_events_override

        # on Insights Cloud, use the feature flag
        if is_cloud():
            return feature_enabled_or_false(
                "persons-on-events-person-id-no-override-properties-on-events",
                str(self.uuid),
                groups={"project": str(self.id)},
                group_properties={
                    "project": {
                        "id": str(self.id),
                        "created_at": self.created_at.isoformat() if self.created_at else None,
                        "uuid": self.uuid,
                    }
                },
                only_evaluate_locally=True,
                send_feature_flag_events=False,
            )

        # on self-hosted, use the instance setting
        return get_instance_setting("PERSON_ON_EVENTS_ENABLED")

    @property
    def _person_on_events_person_id_override_properties_on_events(self) -> bool:
        person_on_events_v2_override = getattr(settings, "PERSON_ON_EVENTS_V2_OVERRIDE", None)
        if person_on_events_v2_override is not None:
            return person_on_events_v2_override

        # on Insights Cloud, use the feature flag
        if is_cloud():
            return feature_enabled_or_false(
                "persons-on-events-v2-reads-enabled",
                str(self.uuid),
                groups={"organization": str(self.organization_id)},
                group_properties={
                    "organization": {
                        "id": str(self.organization_id),
                        "created_at": self.organization.created_at.isoformat()
                        if self.organization.created_at
                        else None,
                    }
                },
                only_evaluate_locally=True,
                send_feature_flag_events=False,
            )

        return get_instance_setting("PERSON_ON_EVENTS_V2_ENABLED")

    @property
    def strict_caching_enabled(self) -> bool:
        enabled_teams = get_list(get_instance_setting("STRICT_CACHING_TEAMS"))
        return str(self.pk) in enabled_teams or "all" in enabled_teams

    @cached_property
    def persons_seen_so_far(self) -> int:
        from insights.datastore.client import sync_execute
        from insights.queries.person_query import PersonQuery

        filter = Filter(data={"full": "true"})
        person_query, person_query_params = PersonQuery(filter, self.id).get_query()

        with tags_context(product=Product.FEATURE_FLAGS, feature=Feature.QUERY):
            return sync_execute(
                f"""
                SELECT count(1) FROM (
                    {person_query}
                )
            """,
                {**person_query_params, **filter.insightsql_context.values},
            )[0][0]

    @lru_cache(maxsize=5)  # noqa: B019 - TODO: refactor to module-level cache
    def groups_seen_so_far(self, group_type_index: GroupTypeIndex) -> int:
        from insights.datastore.client import sync_execute

        with tags_context(product=Product.FEATURE_FLAGS, feature=Feature.QUERY):
            # nosemgrep: datastore-fstring-param-audit - no interpolation, only parameterized values
            return sync_execute(
                f"""
                SELECT
                    count(DISTINCT group_key)
                FROM groups
                WHERE team_id = %(team_id)s AND group_type_index = %(group_type_index)s
            """,
                {"team_id": self.pk, "group_type_index": group_type_index},
            )[0][0]

    @property
    def timezone_info(self) -> ZoneInfo:
        return ZoneInfo(self.timezone)

    def path_cleaning_filter_models(self) -> list["PathCleaningFilter"]:
        from insights.schema import PathCleaningFilter  # noqa: PLC0415

        filters = []
        for f in self.path_cleaning_filters:
            try:
                filters.append(PathCleaningFilter.model_validate(f))
            except pydantic.ValidationError:
                continue
        return filters

    def set_token_and_save(self, *, new_token: str, user: "User", is_impersonated_session: bool):
        new_token = new_token.strip()
        if not new_token:
            raise ValueError("New API token must be non-empty.")
        api_token_field = self._meta.get_field("api_token")
        if not isinstance(api_token_field, models.CharField) or api_token_field.max_length is None:
            raise RuntimeError("Team.api_token field must declare a max_length")
        api_token_max_length = api_token_field.max_length
        if len(new_token) > api_token_max_length:
            raise ValueError(f"New API token exceeds {api_token_max_length} characters.")
        if new_token == self.api_token:
            raise ValueError("New API token is identical to the current token.")
        if Team.objects.filter(api_token=new_token).exclude(pk=self.pk).exists():
            raise ValueError("New API token is already in use by another team.")

        old_token = self.api_token
        self.api_token = new_token
        self._persist_api_token_change(old_token=old_token, user=user, is_impersonated_session=is_impersonated_session)

    def _persist_api_token_change(self, *, old_token: str, user: "User", is_impersonated_session: bool) -> None:
        from insights.models.activity_logging.activity_log import Change, Detail, log_activity

        self.save()
        set_team_in_cache(old_token, None)
        set_team_in_cache(self.api_token, self)
        log_activity(
            organization_id=self.organization_id,
            team_id=self.pk,
            user=cast("User", user),
            was_impersonated=is_impersonated_session,
            scope="Team",
            item_id=self.pk,
            activity="updated",
            detail=Detail(
                name=str(self.name),
                changes=[
                    Change(
                        type="Team",
                        action="changed",
                        field="api_token",
                        before=old_token,
                        after=self.api_token,
                    )
                ],
            ),
        )

    def rotate_secret_token_and_save(self, *, user: "User", is_impersonated_session: bool):
        from insights.models.activity_logging.activity_log import Change, Detail, log_activity

        # Rotate the tokens
        old_primary_token = self.secret_api_token
        new_token = generate_random_token_secret()
        expired_token = self.secret_api_token_backup
        self.secret_api_token = new_token
        self.secret_api_token_backup = old_primary_token
        self.save()

        set_team_in_cache(new_token, self)
        # Old token needs to continue to work until it's deleted.
        if old_primary_token:
            set_team_in_cache(old_primary_token, self)
        if expired_token:
            # Clear the previous backup token from cache since it's being replaced
            set_team_in_cache(expired_token, None)

        secret_api_token_rotated.send(sender=self.__class__, team=self)

        # Build up the changes.

        masked_old_primary_token = mask_key_value(old_primary_token) if old_primary_token else None

        before = {
            "secret_api_token": masked_old_primary_token,
        }
        after = {
            "secret_api_token": mask_key_value(new_token),
        }

        if masked_old_primary_token:
            # We rotated keys rather than generated a new one.
            before["secret_api_token_backup"] = mask_key_value(expired_token) if expired_token else None
            after["secret_api_token_backup"] = masked_old_primary_token

        log_activity(
            organization_id=self.organization_id,
            team_id=self.pk,
            user=cast("User", user),
            was_impersonated=is_impersonated_session,
            scope="Team",
            item_id=self.pk,
            activity="updated",
            detail=Detail(
                name=str(self.name),
                changes=[
                    Change(
                        type="Team",
                        action="created" if old_primary_token is None else "changed",
                        field="secret_api_token",
                        before=before,
                        after=after,
                    )
                ],
            ),
        )

    def generate_conversations_public_token_and_save(self, *, user: "User", is_impersonated_session: bool):
        """Generate or regenerate the conversations public token for widget authentication."""
        from insights.models.activity_logging.activity_log import Change, Detail, log_activity

        settings = self.conversations_settings or {}
        old_token = settings.get("widget_public_token")
        new_token = mint(KeyKind.WIDGET)
        settings["widget_public_token"] = new_token
        self.conversations_settings = settings
        self.save()

        log_activity(
            organization_id=self.organization_id,
            team_id=self.pk,
            user=cast("User", user),
            was_impersonated=is_impersonated_session,
            scope="Team",
            item_id=self.pk,
            activity="updated",
            detail=Detail(
                name=str(self.name),
                changes=[
                    Change(
                        type="Team",
                        action="created" if old_token is None else "changed",
                        field="conversations_settings.widget_public_token",
                        before=mask_key_value(old_token) if old_token else None,
                        after=mask_key_value(new_token),
                    )
                ],
            ),
        )

    def delete_secret_token_backup_and_save(self, *, user: "User", is_impersonated_session: bool):
        from insights.models.activity_logging.activity_log import Change, Detail, log_activity
        from insights.models.utils import mask_key_value

        old_backup_token = self.secret_api_token_backup
        if not old_backup_token:
            # Nothing to delete.
            return

        masked_old_backup_token = mask_key_value(old_backup_token)
        self.secret_api_token_backup = None
        self.save()
        set_team_in_cache(old_backup_token, None)

        log_activity(
            organization_id=self.organization_id,
            team_id=self.pk,
            user=cast("User", user),
            was_impersonated=is_impersonated_session,
            scope="Team",
            item_id=self.pk,
            activity="updated",
            detail=Detail(
                name=str(self.name),
                changes=[
                    Change(
                        type="Team",
                        action="deleted",
                        field="secret_api_token_backup",
                        before=masked_old_backup_token,
                        after=None,
                    )
                ],
            ),
        )

    def get_is_generating_demo_data(self) -> bool:
        cache_key = f"is_generating_demo_data_{self.id}"
        return cache.get(cache_key) == "True"

    def kick_off_demo_data_generation(self, initiating_user: "User") -> None:
        from insights.tasks.demo_create_data import create_data_for_demo_team

        cache_key = f"is_generating_demo_data_{self.id}"
        cache.set(cache_key, "True")  # Create an item in the cache that we can use to see if the demo data is ready
        create_data_for_demo_team.delay(self.id, initiating_user.id, cache_key)

    def all_users_with_access(self) -> QuerySet["User"]:
        # Imported here rather than at module scope because these are the `managed=False` models
        # over the `ee_*` tables, which must not be imported before the app registry is ready. Their
        # uses below outlived the import when `ee/` went, so every call to this raised `NameError` —
        # including the one the request path makes.
        from insights.constants import AvailableFeature
        from insights.models.ee_models import AccessControl, RoleMembership
        from insights.models.organization import OrganizationMembership
        from insights.models.user import User

        # Without ACCESS_CONTROL there is no notion of private teams — all org members have access.
        # Mirrors User.teams and UserTeamPermissions.effective_membership_level_for_parent_membership.
        if not self.organization.is_feature_available(AvailableFeature.ACCESS_CONTROL):
            user_ids_queryset = OrganizationMembership.objects.filter(organization_id=self.organization_id).values_list(
                "user_id", flat=True
            )
            return User.objects.filter(is_active=True, id__in=user_ids_queryset)


        # First, check if the team is private
        team_is_private = AccessControl.objects.filter(
            team_id=self.id,
            resource="project",
            resource_id=str(self.id),
            organization_member=None,
            role=None,
            access_level="none",
        ).exists()

        if not team_is_private:
            # If team is not private, all organization members have access
            user_ids_queryset = OrganizationMembership.objects.filter(organization_id=self.organization_id).values_list(
                "user_id", flat=True
            )
        else:
            # Team is private, need to check specific access

            # Get all organization admins and owners
            admin_user_ids = OrganizationMembership.objects.filter(
                organization_id=self.organization_id,
                level__gte=OrganizationMembership.Level.ADMIN,
            ).values_list("user_id", flat=True)

            # Get users with specific access control entries for this team
            # First, get organization memberships with access to this team
            org_memberships_with_access = AccessControl.objects.filter(
                team_id=self.id,
                resource="project",
                resource_id=str(self.id),
                organization_member__isnull=False,
                access_level__in=["member", "admin"],
            ).values_list("organization_member", flat=True)

            # Then get the user IDs from those memberships
            member_access_user_ids = OrganizationMembership.objects.filter(
                id__in=org_memberships_with_access
            ).values_list("user_id", flat=True)

            # Role-backed access only contributes when the org has ROLE_BASED_ACCESS —
            # same gate as the UI's "Roles" block on the project access settings page.
            if self.organization.is_feature_available(AvailableFeature.ROLE_BASED_ACCESS):
                roles_with_access = AccessControl.objects.filter(
                    team_id=self.id,
                    resource="project",
                    resource_id=str(self.id),
                    role__isnull=False,
                    access_level__in=["member", "admin"],
                ).values_list("role", flat=True)

                role_user_ids = (
                    RoleMembership.objects.filter(role_id__in=roles_with_access)
                    .values_list("organization_member__user_id", flat=True)
                    .distinct()
                )
            else:
                # Empty queryset (not a list) so `.union()` keeps working.
                role_user_ids = RoleMembership.objects.none().values_list("organization_member__user_id", flat=True)

            # Union all sets of user IDs
            user_ids_queryset = cast(Any, admin_user_ids).union(member_access_user_ids, role_user_ids)

        return User.objects.filter(is_active=True, id__in=user_ids_queryset)

    def __str__(self):
        if self.name:
            return self.name
        if self.app_urls and self.app_urls[0]:
            return ", ".join(self.app_urls)
        return str(self.pk)

    __repr__ = sane_repr("id", "uuid", "project_id", "name", "api_token")


@mutable_receiver(post_save, sender=Team)
def put_team_in_cache_on_save(sender, instance: Team, **kwargs):
    set_team_in_cache(instance.api_token, instance)


@mutable_receiver(post_delete, sender=Team)
def delete_team_in_cache_on_delete(sender, instance: Team, **kwargs):
    set_team_in_cache(instance.api_token, None)


@mutable_receiver(post_save, sender=Team)
def reevaluate_authorized_urls_health(sender, instance: Team, **kwargs):
    update_fields = kwargs.get("update_fields")
    if update_fields is not None and "app_urls" not in update_fields:
        return

    from insights.tasks.health_checks import evaluate_health_check_for_team

    team_id = instance.id
    transaction.on_commit(lambda: evaluate_health_check_for_team.delay("authorized_urls", team_id))


def check_is_feature_available_for_team(team_id: int, feature_key: str, current_usage: Optional[int] = None):
    available_product_features: Optional[list[dict[str, str]]] = (
        Team.objects.select_related("organization")
        .values_list("organization__available_product_features", flat=True)
        .get(id=team_id)
    )
    if available_product_features is None:
        return False

    for feature in available_product_features:
        if feature.get("key") == feature_key:
            if current_usage is not None and feature.get("limit") is not None:
                return current_usage < int(feature["limit"])
            return True
    return False