insights/products/logs/backend/models.py

from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING

from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models import Value

from insights.models.activity_logging.model_activity import ModelActivityMixin
from insights.models.scoping.root_mixin import TeamScopedRootMixin
from insights.models.team.extensions import register_team_extension_signal
from insights.models.utils import CreatedMetaFields, UpdatedMetaFields, UUIDModel
from insights.utils import generate_short_id

if TYPE_CHECKING:
    from products.logs.backend.alert_state_machine import AlertSnapshot

logger = logging.getLogger(__name__)

# Default log attribute key whose value matches a Insights person's distinct_id. Mirrors
# the convention documented at https://hanzo.ai/docs/logs/link-session-replay: the
# insights-js / insights-react-native SDKs auto-attach `insightsDistinctId` to every log
# they emit, and the docs instruct OTel-emitting backends to set the same key. Customers
# whose pipeline uses a different key can override via the `logs_config` endpoint.
DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEY = "insightsDistinctId"

DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEYS = [DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEY]


def default_logs_distinct_id_attribute_keys() -> list[str]:
    return list(DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEYS)


# Built-in distinct-id attribute key conventions. Mirror of the frontend DISTINCT_ID_KEYS in
# products/logs/frontend/utils.tsx — keep the two in sync. The logs UI renders a value under
# any of these keys as a clickable person link (isDistinctIdKey), so the person Logs tab scopes
# on them too (on top of a team's configured keys), otherwise a log shown as belonging to a
# person would not appear on that person's tab. Literal keys only: the frontend additionally
# matches dot-suffixed variants (e.g. `span.distinct_id`), which an exact attribute filter can't
# express.
DISTINCT_ID_ATTRIBUTE_KEY_CONVENTIONS = [
    "distinct.id",
    "distinct_id",
    "distinctId",
    "distinctID",
    "insightsDistinctId",
    "insightsDistinctID",
    "insights_distinct_id",
    "insights.distinct.id",
    "insights.distinct_id",
]


# Default log attribute keys whose values hold the Insights session ID. `insightsSessionId`
# is the key the insights-js / insights-react-native SDKs auto-attach to every log they
# emit (see https://hanzo.ai/docs/logs/link-session-replay). Ordered: detection checks
# keys in list order and the first match wins. Customers whose pipeline emits the session
# ID under different keys can override via the `logs_config` endpoint.
DEFAULT_LOGS_SESSION_ID_ATTRIBUTE_KEYS = ["insightsSessionId"]


def default_logs_session_id_attribute_keys() -> list[str]:
    return list(DEFAULT_LOGS_SESSION_ID_ATTRIBUTE_KEYS)


class TeamLogsConfig(models.Model):
    # Plain `models.Model` (not `TeamScopedRootMixin`) — log emission and ingestion
    # are per-environment, and so is this config. Inheriting the root-mixin would
    # rewrite writes to the parent project on save, letting a member of one child
    # environment mutate config that affects sibling environments they may not have
    # access to. Mirrors the `TeamExperimentsConfig` precedent.
    team = models.OneToOneField("insights.Team", on_delete=models.CASCADE, primary_key=True)

    # Legacy single-key predecessor of `logs_distinct_id_attribute_keys`, kept in sync
    # with its first entry so pre-plural readers stay coherent. Do not write directly.
    logs_distinct_id_attribute_key = models.CharField(
        max_length=200,
        default=DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEY,
        db_default=DEFAULT_LOGS_DISTINCT_ID_ATTRIBUTE_KEY,
    )

    # Log attribute keys whose values match a Insights person's distinct_id — a log links
    # to a person when any of these attributes equals one of the person's distinct IDs.
    # Used by the person profile Logs tab and the `query-logs` MCP tool to filter logs
    # to a single user without needing per-team prompt engineering.
    logs_distinct_id_attribute_keys = ArrayField(
        models.CharField(max_length=200),
        default=default_logs_distinct_id_attribute_keys,
        db_default=Value("{insightsDistinctId}"),
    )

    # Ordered list of log attribute keys whose values hold the Insights session ID.
    # Detection checks keys in order; the first key with a value wins. Used to link
    # logs to session replay and error tracking sessions.
    logs_session_id_attribute_keys = ArrayField(
        models.CharField(max_length=200),
        default=default_logs_session_id_attribute_keys,
        db_default=Value("{insightsSessionId}"),
    )


register_team_extension_signal(TeamLogsConfig, logger=logger)

# Upper bound on LogsAlertConfiguration.evaluation_periods. Doubles as the per-alert
# cap on retained OK event rows — the N-of-M evaluator never reads more than this many
# non-errored rows per alert, so older OK rows are pruned. Mirrored in the serializer's
# max_value so the two can't drift.
MAX_EVALUATION_PERIODS = 10


class LogsView(CreatedMetaFields, UpdatedMetaFields, UUIDModel):
    team = models.ForeignKey("insights.Team", on_delete=models.CASCADE)
    short_id = models.CharField(max_length=12, blank=True, default=generate_short_id)
    name = models.CharField(max_length=400)
    filters = models.JSONField(default=dict)
    # Display config (LogsColumnConfig[]), separate from filter state. Null = default column set.
    columns = models.JSONField(null=True, default=None)
    pinned = models.BooleanField(default=False)

    class Meta:
        db_table = "logs_logsview"
        unique_together = ("team", "short_id")
        indexes = [
            models.Index(fields=["team_id", "-created_at"], name="logs_view_team_created_idx"),
        ]

    def __str__(self) -> str:
        return f"{self.name} (Team: {self.team})"


class LogsAlertConfiguration(ModelActivityMixin, CreatedMetaFields, UpdatedMetaFields, UUIDModel):
    class State(models.TextChoices):
        NOT_FIRING = "not_firing", "Not firing"
        FIRING = "firing", "Firing"
        PENDING_RESOLVE = "pending_resolve", "Pending resolve"
        ERRORED = "errored", "Errored"
        SNOOZED = "snoozed", "Snoozed"
        BROKEN = "broken", "Broken"

    class ThresholdOperator(models.TextChoices):
        ABOVE = "above", "Above"
        BELOW = "below", "Below"

    team = models.ForeignKey("insights.Team", on_delete=models.CASCADE)
    name = models.CharField(max_length=255)
    enabled = models.BooleanField(default=True)

    # Filter criteria — subset of LogsViewerFilters (excludes dateRange).
    # Expected shape:
    # {
    #     "severityLevels": list[str],
    #     "serviceNames": list[str],
    #     "filterGroup": {...},
    # }
    filters = models.JSONField(default=dict)

    # Threshold
    threshold_count = models.PositiveIntegerField(default=100)
    threshold_operator = models.CharField(
        max_length=10,
        choices=ThresholdOperator,
        default=ThresholdOperator.ABOVE,
    )

    # Window & scheduling
    window_minutes = models.PositiveIntegerField(default=5)
    check_interval_minutes = models.PositiveIntegerField(default=5)

    # State
    state = models.CharField(
        max_length=20,
        choices=State,
        default=State.NOT_FIRING,
    )

    # N-of-M evaluation (AWS CloudWatch naming convention).
    # evaluation_periods = M, datapoints_to_alarm = N
    evaluation_periods = models.PositiveIntegerField(default=1)
    datapoints_to_alarm = models.PositiveIntegerField(default=1)

    # Cooldown & snooze
    cooldown_minutes = models.PositiveIntegerField(default=0)
    snooze_until = models.DateTimeField(null=True, blank=True)

    # Scheduling & tracking
    next_check_at = models.DateTimeField(null=True, blank=True)
    last_notified_at = models.DateTimeField(null=True, blank=True)
    last_checked_at = models.DateTimeField(null=True, blank=True)
    consecutive_failures = models.PositiveIntegerField(default=0)
    first_enabled_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        db_table = "logs_logsalertconfiguration"
        indexes = [
            models.Index(
                fields=["team_id", "next_check_at", "enabled"],
                name="logs_alert_scheduler_idx",
            ),
        ]

    def __str__(self) -> str:
        return f"{self.name} (Team: {self.team})"

    def clear_next_check(self) -> list[str]:
        """Nulls `next_check_at` so the scheduler picks this alert up on the next tick.
        Returns modified fields for `save(update_fields=...)`.
        """
        self.next_check_at = None
        return ["next_check_at"]

    def to_snapshot(self, recent_events_breached: tuple[bool, ...] | None = None) -> AlertSnapshot:
        """Capture the fields the state machine reads for a transition decision.

        `recent_events_breached` lets the caller pass in the M-of-N window directly
        (e.g. derived from a single bucketed CH query). When omitted, falls back to
        reading historical CHECK rows via `get_recent_breaches` — kept for back-compat
        with code paths that haven't switched to the bucketed eval yet.
        """
        from products.logs.backend.alert_state_machine import AlertSnapshot, AlertState

        return AlertSnapshot(
            state=AlertState(self.state),
            evaluation_periods=self.evaluation_periods,
            datapoints_to_alarm=self.datapoints_to_alarm,
            cooldown_minutes=self.cooldown_minutes,
            last_notified_at=self.last_notified_at,
            snooze_until=self.snooze_until,
            consecutive_failures=self.consecutive_failures,
            recent_events_breached=recent_events_breached
            if recent_events_breached is not None
            else self.get_recent_breaches(),
        )

    def get_recent_breaches(self) -> tuple[bool, ...]:
        """Last M non-errored check events' threshold_breached values, newest first."""
        return tuple(
            LogsAlertEvent.objects.filter(
                alert=self,
                kind=LogsAlertEvent.Kind.CHECK,
                error_message__isnull=True,
            )
            .order_by("-created_at")
            .values_list("threshold_breached", flat=True)[: self.evaluation_periods]
        )

    def clean(self) -> None:
        super().clean()
        if self.datapoints_to_alarm > self.evaluation_periods:
            raise ValidationError(
                f"datapoints_to_alarm cannot exceed evaluation_periods ({self.datapoints_to_alarm} > {self.evaluation_periods})"
            )


class LogsAlertCheck(UUIDModel):
    """Defunct — kept in sync with the physical table `logs_logsalertcheck`.

    All production reads and writes go through `LogsAlertEvent` (the new table). This
    shell class exists solely to match Django's model state with the legacy table
    created by `0001_initial.py`. PR 4 will drop the table and remove this class.
    """

    alert = models.ForeignKey(
        LogsAlertConfiguration,
        on_delete=models.CASCADE,
        related_name="checks",
    )
    created_at = models.DateTimeField(auto_now_add=True)
    result_count = models.PositiveIntegerField(null=True, blank=True)
    threshold_breached = models.BooleanField()
    state_before = models.CharField(max_length=20)
    state_after = models.CharField(max_length=20)
    error_message = models.TextField(null=True, blank=True)
    query_duration_ms = models.PositiveIntegerField(null=True, blank=True)

    class Meta:
        db_table = "logs_logsalertcheck"


class LogsAlertEvent(UUIDModel):
    # Events (errored, breached, state-transition rows) retained this long for forensics.
    # OK rows are capped by count (MAX_EVALUATION_PERIODS per alert) rather than by time.
    EVENT_RETENTION_DAYS = 90

    class Kind(models.TextChoices):
        # Worker-produced row from evaluating the Datastore check query. Only CHECK rows
        # feed the N-of-M evaluator and are eligible for the inline prune. Control-plane
        # kinds are reserved for user-initiated state transitions; writers are added in a
        # follow-up PR (see spike 4.7). Every read path must filter by kind=CHECK to keep
        # control-plane rows out of evaluator and prune windows.
        CHECK = "check", "Check"
        RESET = "reset", "Reset"
        ENABLE = "enable", "Enable"
        DISABLE = "disable", "Disable"
        SNOOZE = "snooze", "Snooze"
        UNSNOOZE = "unsnooze", "Unsnooze"
        THRESHOLD_CHANGE = "threshold_change", "Threshold change"
        BROKEN_CONFIG = "broken_config", "Broken config"

    alert = models.ForeignKey(
        LogsAlertConfiguration,
        on_delete=models.CASCADE,
        related_name="events",
    )
    kind = models.CharField(max_length=32, choices=Kind.choices, default=Kind.CHECK)
    created_at = models.DateTimeField(auto_now_add=True)
    result_count = models.PositiveIntegerField(null=True, blank=True)
    threshold_breached = models.BooleanField()
    state_before = models.CharField(max_length=20)
    state_after = models.CharField(max_length=20)
    error_message = models.TextField(null=True, blank=True)
    query_duration_ms = models.PositiveIntegerField(null=True, blank=True)

    class Meta:
        db_table = "logs_logsalertevent"
        indexes = [
            models.Index(fields=["alert", "-created_at"], name="logs_alert_event_alert_ts_idx"),
        ]

    def __str__(self) -> str:
        return f"LogsAlertEvent for {self.alert.name} at {self.created_at}"

    @classmethod
    def clean_up_old_events(cls) -> int:
        """Delete every event row older than EVENT_RETENTION_DAYS.

        In steady state this only touches errored rows and state-transition rows: the
        Temporal activity caps non-event rows to MAX_EVALUATION_PERIODS per alert
        inline. Rows from silent or disabled alerts also age out through this path.
        """
        oldest = datetime.now(UTC) - timedelta(days=cls.EVENT_RETENTION_DAYS)
        count, _ = cls.objects.filter(created_at__lt=oldest).delete()
        return count


# Cap on enabled metric rules per team. Every enabled rule is evaluated against every
# ingested log record in the Node worker, so the cap bounds per-record CPU. Mirrored in
# the serializer so the limit surfaces as a 400 rather than silent worker truncation.
MAX_ENABLED_METRIC_RULES = 10

# Group-by cardinality bounds. Keys beyond the cap multiply the number of emitted metric
# series per rule; the serializer rejects rules exceeding it at write time.
MAX_METRIC_RULE_GROUP_BY_KEYS = 5

# Top-level LogRecord fields allowed as group-by dimensions. Anything else must be
# addressed through the `attributes.` / `resource_attributes.` map prefixes.
METRIC_RULE_GROUP_BY_TOP_LEVEL_KEYS = ("service_name", "severity_text", "event_name")


class LogsMetricRule(ModelActivityMixin, TeamScopedRootMixin, CreatedMetaFields, UpdatedMetaFields, UUIDModel):
    """Generates a metric from ingested logs: records matching `filter_group` are tallied
    at ingest time (before drop rules) and emitted into the Metrics product under
    `metric_name`. With `value_attribute` unset the rule counts matching records; when set,
    the numeric value of that attribute is aggregated into a distribution (count + sum)."""

    # db_constraint=False on the team/user FKs: insights_team and insights_user are hot tables,
    # and creating an FK constraint against them locks the parent — see the hot-table section
    # of safe-django-migrations.md. Enforcement is app-level (Django still cascades in the ORM).
    team = models.ForeignKey("insights.Team", on_delete=models.CASCADE, db_constraint=False)
    created_by = models.ForeignKey(
        "insights.User", on_delete=models.SET_NULL, null=True, blank=True, related_name="+", db_constraint=False
    )
    name = models.CharField(max_length=255)
    # Emitted OTLP metric name. Immutable after create (changing it would start a brand-new
    # series and orphan the old one) — enforced in the serializer.
    metric_name = models.CharField(max_length=200)
    enabled = models.BooleanField(default=False)
    # PropertyGroupFilter JSON (same shape as LogsExclusionRule config.filter_group).
    # Null = every ingested log record matches.
    filter_group = models.JSONField(null=True, blank=True, default=None)
    # Log attribute key holding the numeric value to aggregate (`attributes.` /
    # `resource_attributes.` prefixed). Null = count matching records. Immutable after
    # create — it decides the emitted metric type (sum vs histogram).
    value_attribute = models.CharField(max_length=512, null=True, blank=True)
    # Group-by dimension keys; each distinct value combination becomes its own series.
    group_by = ArrayField(models.CharField(max_length=512), default=list, blank=True)
    version = models.PositiveIntegerField(default=1)

    class Meta:
        db_table = "logs_logsmetricrule"
        constraints = [
            models.UniqueConstraint(fields=["team", "metric_name"], name="logs_metric_rule_team_metric_uniq"),
        ]
        indexes = [
            models.Index(fields=["team_id", "enabled"], name="logs_metric_team_enabled_idx"),
        ]

    def __str__(self) -> str:
        return f"{self.name} -> {self.metric_name} (team={self.team_id})"


class LogsExclusionRule(ModelActivityMixin, CreatedMetaFields, UpdatedMetaFields, UUIDModel):
    """User-defined rules to drop or exclude log lines before storage (evaluated in ingestion when enabled)."""

    class RuleType(models.TextChoices):
        SEVERITY_SAMPLING = "severity_sampling", "Severity-based reduction"
        PATH_DROP = "path_drop", "Path exclusion"
        RATE_LIMIT = "rate_limit", "Rate limit"

    team = models.ForeignKey("insights.Team", on_delete=models.CASCADE)
    name = models.CharField(max_length=255)
    enabled = models.BooleanField(default=False)
    priority = models.PositiveIntegerField(
        default=0,
        help_text="Lower values run first; first matching rule wins. Ties use created_at ascending (same as ingestion query order).",
    )
    rule_type = models.CharField(max_length=32, choices=RuleType.choices)
    scope_service = models.CharField(max_length=512, null=True, blank=True)
    scope_path_pattern = models.CharField(max_length=1024, null=True, blank=True)
    scope_attribute_filters = models.JSONField(default=list)
    config = models.JSONField(default=dict)
    version = models.PositiveIntegerField(default=1)

    class Meta:
        db_table = "logs_logsexclusionrule"
        indexes = [
            models.Index(fields=["team_id", "enabled", "priority"], name="logs_exclusion_team_en_pr_idx"),
        ]

    def __str__(self) -> str:
        return f"{self.name} (team={self.team_id})"


class LogsRetentionRule(ModelActivityMixin, CreatedMetaFields, UpdatedMetaFields, UUIDModel):
    """User-defined rules that override how long matching log lines are retained (evaluated in ingestion
    when enabled). First matching rule by (priority, created_at) wins; logs matching no rule keep the
    team's default retention (`Team.logs_settings.retention_days`)."""

    # Plain team FK — like LogsExclusionRule and TeamLogsConfig, retention rules are per-environment,
    # so this deliberately does not use TeamScopedRootMixin (whose canonical-team save() rewrite would
    # let one child environment mutate a sibling's rules). Tenant isolation is enforced at the API layer
    # via safely_get_queryset filtering on team_id; the model is tracked in scoping/baseline_unmigrated.txt.
    # db_constraint=False on the hot-table FKs (team, created_by) keeps the CreateModel migration
    # lock-free — creating a real FK constraint would take a SHARE ROW EXCLUSIVE lock on the parent.
    # Enforcement stays at the ORM level (cascade/set-null run through the Django collector).
    team = models.ForeignKey("insights.Team", on_delete=models.CASCADE, db_constraint=False)
    created_by = models.ForeignKey(
        "insights.User", on_delete=models.SET_NULL, null=True, blank=True, db_constraint=False
    )
    name = models.CharField(max_length=255)
    enabled = models.BooleanField(default=False)
    priority = models.PositiveIntegerField(
        default=0,
        help_text="Lower values run first; first matching rule wins. Ties use created_at ascending (same as ingestion query order).",
    )
    # {"filter_group": <PropertyGroupFilter>, "retention_days": <14|30|90>}
    config = models.JSONField(default=dict)
    version = models.PositiveIntegerField(default=1)

    class Meta:
        db_table = "logs_logsretentionrule"
        indexes = [
            models.Index(fields=["team_id", "enabled", "priority"], name="logs_retention_team_en_pr_idx"),
        ]

    def __str__(self) -> str:
        return f"{self.name} (team={self.team_id})"