insights/products/conversations/backend/events.py

"""
Emit Insights analytics events for ticket and message state changes.

These events power workflow triggers (e.g. "if ticket pending for X days, resolve it").
Staff/system actions use the actor's distinct_id; customer actions use the ticket's distinct_id.
Events are sent to the customer's Insights project via their team's API token.
"""

from datetime import datetime
from typing import Literal

from django.utils import timezone

import structlog

from insights.api.capture import capture_internal
from insights.datastore.query_tagging import Feature, Product, tags_context
from insights.event_usage import groups as build_groups
from insights.models.group.util import get_groups_by_identifiers
from insights.models.group_type_mapping import get_group_types_for_project
from insights.models.organization import OrganizationMembership
from insights.models.person.person import Person
from insights.models.person.util import get_persons_by_distinct_ids
from insights.models.team import Team
from insights.models.user import User
from insights.personinsights_client.caller_tag import personinsights_caller_tag
from insights.settings import SITE_URL

from products.conversations.backend.cache import get_cached_resolved_groups, set_cached_resolved_groups
from products.conversations.backend.models import Ticket, TicketAssignment
from products.conversations.backend.models.constants import Channel, OrganizationIdSource

from insights.models.ee_models import Role

logger = structlog.get_logger(__name__)

EVENT_SOURCE = "conversations_events"

ActorType = Literal["user", "system", "external", "customer"]


def _get_actor_properties(actor: User | None, actor_type: ActorType) -> dict:
    """Build actor-related properties for an event."""
    return {
        "actor_type": actor_type,
        "actor_id": actor.id if actor else None,
        "actor_email": actor.email if actor else None,
    }


def _get_actor_distinct_id(
    ticket: Ticket,
    actor: User | None,
    actor_type: ActorType,
) -> str:
    """Determine the distinct_id based on who performed the action.

    User actions are attributed to the staff member.
    System/external/customer actions are attributed to the customer.
    """
    if actor_type == "user" and actor and actor.distinct_id:
        return actor.distinct_id
    return ticket.distinct_id or ticket.channel_source or "unknown"


# Channels whose customer email is tied to a provider-verified identity and is therefore safe
# to use for organization attribution.
_EMAIL_FALLBACK_CHANNELS = frozenset({Channel.EMAIL.value, Channel.SLACK.value, Channel.TEAMS.value})

# Latest organization/customer group keys from the customer's recent events. We read the
# dedicated $group_N columns (set on every event that carries $groups) rather than filtering on
# $groupidentify events: insights.group(type, key) called without properties associates the group
# with all subsequent events but emits no $groupidentify (the documented SDK behavior), and even
# newer SDKs only re-emit it for newly-seen groups. A $groupidentify filter would therefore
# silently miss exactly the cross-region customers this fallback exists for — those whose apps
# don't pass group properties, or whose last group identify predates the 30-day window.
# {org_col}/{customer_select} are interpolated from this project's own group-type indexes (see
# _resolve_groups_from_analytics); column names can't be InsightsQL placeholders.
GROUPS_FROM_EVENTS_QUERY = """
SELECT
    argMax({org_col}, timestamp),
    {customer_select}
FROM events
WHERE distinct_id IN {{distinct_ids}}
  AND timestamp >= now() - INTERVAL 30 DAY
  AND {org_col} != ''
"""


def _resolve_groups_from_analytics(team: Team, distinct_ids: list[str]) -> dict | None:
    """Resolve ``$groups`` from the customer's recent analytics events in Datastore.

    Fallback for when the ``OrganizationMembership`` lookup misses: that join runs
    against this region's Postgres only, so accounts registered in another region
    (e.g. an EU user filing a ticket against a US project) are invisible to it.
    Their identified app sessions, however, stamp ``$group_N`` onto every event
    captured into this project, which is cross-region by construction.

    Event-supplied groups are captured with the project's public token and are
    therefore spoofable — fine for analytics enrichment (same trust level as
    ``$identify``), never for authorization. ``instance``/``project`` are rebuilt
    server-side so fallback-path events match ``build_groups()`` output.
    """
    if not distinct_ids:
        return None

    cached = get_cached_resolved_groups(team.id, distinct_ids)
    if cached is not None:
        return cached or None  # {} is the negative-cache sentinel

    # The $group_N column index is per-project, so resolve the org/customer indexes from this
    # project's group-type mapping. Doubles as a cheap guard: bail (and negative-cache) for
    # projects without an organization group type before touching Datastore.
    group_type_index = {
        gtm["group_type"]: gtm["group_type_index"] for gtm in get_group_types_for_project(team.project_id)
    }
    org_index = group_type_index.get("organization")
    if org_index is None:
        set_cached_resolved_groups(team.id, distinct_ids, None)
        return None
    customer_index = group_type_index.get("customer")

    # Indexes are trusted ints (0-4) from the project's own mapping; safe to interpolate.
    org_col = f"`$group_{org_index}`"
    customer_select = (
        f"argMaxIf(`$group_{customer_index}`, timestamp, `$group_{customer_index}` != '')"
        if customer_index is not None
        else "''"
    )
    query = GROUPS_FROM_EVENTS_QUERY.format(org_col=org_col, customer_select=customer_select)

    # Deferred: insightsql.query pulls the whole query-runner layer, and this module loads
    # at django.setup() via the conversations signal wiring.
    from insights.insightsql import ast  # noqa: PLC0415
    from insights.insightsql.query import execute_insightsql_query  # noqa: PLC0415

    with tags_context(product=Product.CONVERSATIONS, feature=Feature.QUERY):
        response = execute_insightsql_query(
            query,
            placeholders={"distinct_ids": ast.Constant(value=distinct_ids)},
            team=team,
            query_type="conversations_groups_lookup",
        )

    groups: dict | None = None
    if response.results:
        org_key, customer_key = response.results[0]
        if org_key:
            groups = {"instance": SITE_URL, "project": str(team.uuid), "organization": org_key}
            if customer_key:
                groups["customer"] = customer_key

    set_cached_resolved_groups(team.id, distinct_ids, groups)
    return groups


def _resolve_groups_from_person_properties(team: Team, person: Person) -> dict | None:
    """Resolve ``$groups`` from the person's own ``organization_id`` profile property.

    Fallback for when both the membership lookup and the analytics fallback miss:
    app instrumentation commonly stamps the org id onto the person profile via
    ``$identify``, and the profile persists even when the person's last identified
    session predates the analytics fallback's 30-day event window.

    Person properties are client-supplied — the same trust level as the analytics
    fallback (enrichment, never authorization). The value only counts when it names
    an organization group that already exists in this project, so a team whose
    ``organization_id`` person property follows a different convention than its
    organization group keys can't stamp bogus group keys onto events. A value that
    coincidentally matches an unrelated org group key (e.g. small-integer keys) can
    still mis-enrich — same collision class as event-supplied ``$groups``.
    """
    org_id = (person.properties or {}).get("organization_id")
    if not org_id or not isinstance(org_id, str):
        return None

    group_type_index = {
        gtm["group_type"]: gtm["group_type_index"] for gtm in get_group_types_for_project(team.project_id)
    }
    org_index = group_type_index.get("organization")
    if org_index is None:
        return None
    if not get_groups_by_identifiers(team.id, org_index, [org_id]):
        return None

    return {"instance": SITE_URL, "project": str(team.uuid), "organization": org_id}


def _org_groups_for_person(ticket: Ticket, team: Team, person: Person, distinct_ids: list[str]) -> dict | None:
    """Resolve org ``$groups`` for an already-resolved person: membership, then analytics
    events, then the person's own profile.

    Each lookup hits a different subsystem (Postgres, Datastore, personinsights), so failures
    are isolated per step: the created event is emitted exactly once, and a transient error
    in one source must degrade to the next fallback rather than permanently losing the
    ticket's attribution (which downstream SLA/tagging workflows key off).
    """
    if distinct_ids:
        try:
            membership = (
                OrganizationMembership.objects.select_related("organization")
                .filter(user__distinct_id__in=distinct_ids)
                .first()
            )
            if membership:
                return build_groups(membership.organization, team)
        except Exception:
            logger.exception("ticket_org_membership_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))
        # Membership rows are region-local: accounts registered in another region
        # never appear in this Postgres. Fall back to the $groups their sessions
        # stamped onto this project's analytics events.
        try:
            groups = _resolve_groups_from_analytics(team, distinct_ids)
            if groups:
                return groups
        except Exception:
            logger.exception("ticket_org_analytics_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))
    # Their events may have aged out of the analytics window; the profile they
    # identified with is still the same person's own claim about their org.
    try:
        return _resolve_groups_from_person_properties(team, person)
    except Exception:
        logger.exception("ticket_org_person_properties_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))
        return None


def _resolve_person_org_groups(ticket: Ticket, team: Team) -> tuple[bool, dict | None]:
    """Resolve the customer organization's ``$groups`` from the requester's identity.

    Returns ``(process_person, groups)``. ``groups`` is ``None`` when the
    customer can't be tied to a Insights organization.

    The web widget sets ``ticket.distinct_id`` to a real Insights distinct_id,
    so the membership is found directly. Other channels (Slack, Email, MS
    Teams) set ``distinct_id`` to the customer email (or empty), so we fall
    back to looking the person up by ``properties.email`` in Datastore and
    resolving the membership through that person's real distinct_ids. Either
    way, the person's own ``organization_id`` profile property is the last
    resort when the membership and analytics lookups both miss.
    """
    # 1. Real distinct_id (web widget). An identified person is authoritative: if they
    # have no org membership, don't guess via email (a shared email could resolve to a
    # different person's org), so return early instead of falling through.
    if ticket.distinct_id:
        persons: list[Person] = []
        try:
            # Only is_identified and properties are read, and the membership lookup keys off
            # the ticket's own distinct_id — so skip fetching the person's distinct_ids.
            with personinsights_caller_tag("conversations/ticket-event-person"):
                persons = get_persons_by_distinct_ids(team.id, [ticket.distinct_id], distinct_id_limit=0)
        except Exception:
            # A failed lookup is not "no identified person": fall through to the email
            # path instead of losing attribution to a transient personinsights error.
            logger.exception("ticket_org_person_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))
        identified_person = next((p for p in persons if p.is_identified), None)
        if identified_person is not None:
            groups = _org_groups_for_person(ticket, team, identified_person, [ticket.distinct_id])
            if groups:
                return True, groups
            return False, None

    # 2. Email fallback, restricted to channels with a provider-verified email identity.
    # Never trust the public widget's attacker-controlled anonymous_traits.email here.
    if ticket.channel_source not in _EMAIL_FALLBACK_CHANNELS:
        return False, None

    email = (ticket.anonymous_traits or {}).get("email") or ticket.email_from
    if email:
        # person_lookup pulls the InsightsQL query layer; this module loads at django.setup()
        # via the conversations signal wiring, so import it lazily.
        from products.conversations.backend.person_lookup import _get_persons_by_email  # noqa: PLC0415

        person: Person | None = None
        try:
            person = _get_persons_by_email(team, [email]).get(email.lower())
        except Exception:
            # Same isolation as above: a failed email lookup still leaves the
            # Slack-channel fallback in _resolve_org_groups a chance to attribute.
            logger.exception("ticket_org_email_person_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))
        if person is not None:
            groups = _org_groups_for_person(ticket, team, person, person.distinct_ids or [])
            if groups:
                return True, groups

    return False, None


def _resolve_groups_from_slack_channel(team: Team, slack_channel_id: str) -> dict | None:
    """Attribute the ticket to the customer analytics account bound to its Slack channel.

    Last-resort fallback when the requester can't be tied to an organization: a
    dedicated support channel maps to one account, and that account's ``external_id``
    is the group key at the team's configured account group type. Channel-level
    evidence only — it says which account the ticket belongs to, not who sent it.
    """
    group_type_index = team.customer_analytics_config.account_group_type_index
    if group_type_index is None:
        return None

    group_type_name = next(
        (
            gtm["group_type"]
            for gtm in get_group_types_for_project(team.project_id)
            if gtm["group_type_index"] == group_type_index
        ),
        None,
    )
    if group_type_name is None:
        return None

    # The facade pulls the workflows/notebooks facades with it; keep that off the
    # django.setup() path this module loads on via the conversations signal wiring.
    from products.customer_analytics.backend.facade import api as customer_analytics  # noqa: PLC0415

    account = customer_analytics.get_account_ref_by_slack_channel_id(team.id, slack_channel_id)
    if account is None or not account.external_id:
        return None

    return {"instance": SITE_URL, "project": str(team.uuid), group_type_name: account.external_id}


def _resolve_org_groups(ticket: Ticket, team: Team) -> tuple[bool, dict | None, str | None]:
    """Resolve a ticket's ``$groups``, preferring the requester's identity.

    Returns ``(process_person, groups, source)`` where ``source`` is an
    ``OrganizationIdSource`` value recording how ``groups`` was resolved, or
    ``None`` when nothing resolved. Slack tickets that can't be attributed
    through the person fall back to the customer analytics account associated
    with the ticket's Slack channel.
    """
    process_person, groups = _resolve_person_org_groups(ticket, team)
    if groups is not None:
        return process_person, groups, OrganizationIdSource.PERSON

    # Deliberately no analytics fallback keyed on the bare ticket distinct_id here: for
    # email/Slack/Teams that value is the raw customer email, and events under it are
    # captured with the public project token, so anyone who knows the email can plant a
    # $groups value. identity_verified only attests the ticket's sender, not who captured
    # historical events under that key. Event-derived attribution is only trusted when
    # anchored to a resolved person (see _org_groups_for_person).
    if ticket.channel_source == Channel.SLACK.value and ticket.slack_channel_id:
        groups = _resolve_groups_from_slack_channel(team, ticket.slack_channel_id)
        if groups is not None:
            return process_person, groups, OrganizationIdSource.SLACK_CHANNEL_ACCOUNT

    return process_person, None, None


def _groups_from_org_id(team: Team, organization_id: str) -> dict:
    """Rebuild minimal $groups from a stored org id, skipping the expensive resolver."""
    return {"instance": SITE_URL, "project": str(team.uuid), "organization": organization_id}


def _get_ticket_base_properties(ticket: Ticket) -> dict:
    return {
        "ticket_id": str(ticket.id),
        "ticket_number": ticket.ticket_number,
        "channel_source": ticket.channel_source,
        "channel_detail": ticket.channel_detail,
        "status": ticket.status,
        "priority": ticket.priority,
    }


def _get_customer_properties(ticket: Ticket, *, include_distinct_id: bool = False) -> dict:
    """Customer identity on the ticket, for workflow filters and analytics."""
    traits = ticket.anonymous_traits or {}
    properties = {
        "customer_name": traits.get("name", ""),
        "customer_email": traits.get("email") or ticket.email_from or "",
    }
    if include_distinct_id:
        properties["customer_distinct_id"] = ticket.distinct_id or ""
    return properties


def _get_assignment_properties(ticket: Ticket) -> dict:
    """Current assignment on the ticket, so workflows can route a single team's tickets.

    `assignee_type` is "user", "role", or None (unassigned). `assignee_role_name` is set only for
    role assignments, letting filters target a team by name instead of its UUID.
    """
    assignment = TicketAssignment.objects.select_related("role").filter(ticket_id=ticket.id).first()
    if assignment is None:
        return {"assignee_type": None, "assignee_id": None, "assignee_role_name": None}
    if assignment.user_id is not None:
        return {"assignee_type": "user", "assignee_id": str(assignment.user_id), "assignee_role_name": None}
    if assignment.role_id is not None:
        return {
            "assignee_type": "role",
            "assignee_id": str(assignment.role_id),
            "assignee_role_name": assignment.role.name if assignment.role else None,
        }
    return {"assignee_type": None, "assignee_id": None, "assignee_role_name": None}


def _role_name_for_assignee(ticket: Ticket, assignee_type: str | None, assignee_id: str | None) -> str | None:
    """Look up a role's name for the assignment event; None for user or empty assignees."""
    if assignee_type != "role" or not assignee_id:
        return None
    return (
        Role.objects.filter(id=assignee_id, organization_id=ticket.team.organization_id)
        .values_list("name", flat=True)
        .first()
    )


def _get_sla_properties(ticket: Ticket, now: datetime) -> dict:
    """SLA state at the moment of the event.

    Stamped on events (rather than derived later) so attainment metrics reflect the
    deadline that was in force at the time, even if the SLA is reset afterwards.
    `sla_delta_seconds` is positive when past due, negative when time remains.
    """
    if ticket.sla_due_at is None:
        return {"sla_due_at": None, "sla_active": False, "sla_breached": False, "sla_delta_seconds": None}
    delta_seconds_float = (now - ticket.sla_due_at).total_seconds()
    return {
        "sla_due_at": ticket.sla_due_at.isoformat(),
        "sla_active": True,
        "sla_breached": delta_seconds_float > 0,
        "sla_delta_seconds": int(delta_seconds_float),
    }


def capture_ticket_created(ticket: Ticket) -> None:
    properties = _get_ticket_base_properties(ticket)
    properties.update(_get_customer_properties(ticket))

    team = ticket.team
    team_id = team.id
    process_person = False
    try:
        process_person, groups, groups_source = _resolve_org_groups(ticket, team)
        if groups is not None:
            properties["$groups"] = groups
            org_id = groups.get("organization")
            if org_id and not ticket.organization_id:
                Ticket.objects.filter(id=ticket.id).update(organization_id=org_id, organization_id_source=groups_source)
                ticket.organization_id = org_id
                ticket.organization_id_source = groups_source
    except Exception:
        logger.exception("ticket_created_person_lookup_failed", team_id=team_id, ticket_id=str(ticket.id))

    capture_internal(
        token=team.api_token,
        event_name="$conversation_ticket_created",
        event_source=EVENT_SOURCE,
        distinct_id=ticket.distinct_id or ticket.channel_source or "unknown",
        timestamp=None,
        properties=properties,
        process_person_profile=process_person,
    )


def capture_ticket_status_changed(
    ticket: Ticket,
    old_status: str,
    new_status: str,
    actor: User | None = None,
    actor_type: ActorType = "system",
) -> None:
    properties = _get_ticket_base_properties(ticket)
    properties["old_status"] = old_status
    properties["new_status"] = new_status
    properties.update(_get_actor_properties(actor, actor_type))
    properties.update(_get_customer_properties(ticket, include_distinct_id=True))

    capture_internal(
        token=ticket.team.api_token,
        event_name="$conversation_ticket_status_changed",
        event_source=EVENT_SOURCE,
        distinct_id=_get_actor_distinct_id(ticket, actor, actor_type),
        timestamp=None,
        properties=properties,
    )


def capture_ticket_priority_changed(
    ticket: Ticket,
    old_priority: str | None,
    new_priority: str | None,
    actor: User | None = None,
    actor_type: ActorType = "system",
) -> None:
    properties = _get_ticket_base_properties(ticket)
    properties["old_priority"] = old_priority
    properties["new_priority"] = new_priority
    properties.update(_get_actor_properties(actor, actor_type))
    properties.update(_get_customer_properties(ticket, include_distinct_id=True))

    capture_internal(
        token=ticket.team.api_token,
        event_name="$conversation_ticket_priority_changed",
        event_source=EVENT_SOURCE,
        distinct_id=_get_actor_distinct_id(ticket, actor, actor_type),
        timestamp=None,
        properties=properties,
    )


def capture_ticket_assigned(
    ticket: Ticket,
    assignee_type: str | None,
    assignee_id: str | None,
    actor: User | None = None,
    actor_type: ActorType = "system",
) -> None:
    properties = _get_ticket_base_properties(ticket)
    properties["assignee_type"] = assignee_type
    properties["assignee_id"] = assignee_id
    properties["assignee_role_name"] = _role_name_for_assignee(ticket, assignee_type, assignee_id)
    properties.update(_get_actor_properties(actor, actor_type))
    properties.update(_get_customer_properties(ticket, include_distinct_id=True))

    capture_internal(
        token=ticket.team.api_token,
        event_name="$conversation_ticket_assigned",
        event_source=EVENT_SOURCE,
        distinct_id=_get_actor_distinct_id(ticket, actor, actor_type),
        timestamp=None,
        properties=properties,
    )


def _capture_team_message(
    ticket: Ticket,
    message_id: str,
    message_content: str | None,
    author: User | None,
    event_name: str,
) -> None:
    properties = _get_ticket_base_properties(ticket)
    properties["message_id"] = message_id
    if message_content is not None:
        properties["message_content"] = (message_content or "")[:1000]
    properties["author_type"] = "team"
    properties.update(_get_actor_properties(author, "user"))
    properties.update(_get_customer_properties(ticket, include_distinct_id=True))
    properties.update(_get_assignment_properties(ticket))
    properties.update(_get_sla_properties(ticket, timezone.now()))

    capture_internal(
        token=ticket.team.api_token,
        event_name=event_name,
        event_source=EVENT_SOURCE,
        distinct_id=_get_actor_distinct_id(ticket, author, "user"),
        timestamp=None,
        properties=properties,
    )


def capture_message_sent(
    ticket: Ticket,
    message_id: str,
    message_content: str,
    author: User | None = None,
) -> None:
    """Team member sent a public reply on a ticket."""
    _capture_team_message(ticket, message_id, message_content, author, "$conversation_message_sent")


def capture_private_message_sent(
    ticket: Ticket,
    message_id: str,
    author: User | None = None,
) -> None:
    """Team member sent a private internal note on a ticket.

    Deliberately a separate event from `$conversation_message_sent`: existing
    workflows trigger on that event to notify customers, and private notes must
    never flow through those.

    The note body is deliberately omitted. Analytics events are queryable by any
    project member, so including it would bypass ticket-level access controls.
    """
    _capture_team_message(ticket, message_id, None, author, "$conversation_private_message_sent")


def capture_message_received(ticket: Ticket, message_id: str, message_content: str) -> None:
    """Customer sent a message on a ticket."""
    properties = _get_ticket_base_properties(ticket)
    properties["message_id"] = message_id
    properties["message_content"] = (message_content or "")[:1000]
    properties["author_type"] = "customer"
    properties.update(_get_customer_properties(ticket))
    properties.update(_get_assignment_properties(ticket))

    team = ticket.team
    process_person = False
    try:
        if ticket.organization_id:
            properties["$groups"] = _groups_from_org_id(team, ticket.organization_id)
            # Only a person-resolved org attests the sender (legacy rows without a source
            # were person-resolved); a channel-inferred org says nothing about them, so
            # don't create a person profile for it. Allowlist rather than blocklist so a
            # future non-person source defaults to no person processing.
            process_person = ticket.organization_id_source in (OrganizationIdSource.PERSON, None, "")
        else:
            process_person, groups, groups_source = _resolve_org_groups(ticket, team)
            if groups is not None:
                properties["$groups"] = groups
                org_id = groups.get("organization")
                if org_id:
                    # Persist like capture_ticket_created does, so later messages take the
                    # stored-org fast path instead of re-running the resolver. First write
                    # wins: only mirror to the in-memory ticket when this update landed,
                    # so a concurrent handler's value isn't shadowed.
                    rows_updated = Ticket.objects.filter(id=ticket.id, organization_id__isnull=True).update(
                        organization_id=org_id, organization_id_source=groups_source
                    )
                    if rows_updated:
                        ticket.organization_id = org_id
                        ticket.organization_id_source = groups_source
    except Exception:
        logger.exception("message_received_person_lookup_failed", team_id=team.id, ticket_id=str(ticket.id))

    capture_internal(
        token=team.api_token,
        event_name="$conversation_message_received",
        event_source=EVENT_SOURCE,
        distinct_id=ticket.distinct_id or ticket.channel_source or "unknown",
        timestamp=None,
        properties=properties,
        process_person_profile=process_person,
    )