@snowflake_id_model
@cell_silo_model
class Organization(ReplicatedCellModel):
    """
    An organization represents a group of individuals which maintain ownership of projects.
    """
    category = OutboxCategory.ORGANIZATION_UPDATE
    replication_version = 5

    name = models.CharField(max_length=64)
    slug = SentryOrgSlugField(unique=True)
    status = BoundedPositiveIntegerField(
        choices=OrganizationStatus.as_choices(),
        default=OrganizationStatus.ACTIVE.value
    )
    date_added = models.DateTimeField(default=timezone.now)
    default_role = models.CharField(max_length=32, default=str(roles.get_default().id))
    is_test = models.BooleanField(default=False, db_default=False)

    class flags(TypedClassBitField):
        allow_joinleave: bool
        enhanced_privacy: bool
        disable_shared_issues: bool
        early_adopter: bool
        require_2fa: bool
        disable_new_visibility_features: bool
        # ... 更多位标志字段

class OrganizationManager(BaseManager["Organization"]):
    def get_for_user_ids(self, user_ids):
        """返回一组用户有权访问的所有组织"""
        return self.filter(
            status=OrganizationStatus.ACTIVE,
            member_set__user_id__in=user_ids,
        )

    def get_for_user(self, user, scope=None, only_visible=True):
        """返回单个用户有权访问的组织列表"""
        # 通过 OrganizationMember 中间表关联查询
        qs = OrganizationMember.objects.filter(user_id=user.id).select_related("organization")
        if only_visible:
            qs = qs.filter(organization__status=OrganizationStatus.ACTIVE)
        # ...

class Project(Model):
    """
    Projects are permission based namespaces which generally
    are the top level entry point for all data.
    """
    slug = SentrySlugField(max_length=50)
    name = models.CharField(max_length=200)
    organization = FlexibleForeignKey("sentry.Organization")
    teams = models.ManyToManyField("sentry.Team", related_name="teams",
                                   through=ProjectTeam)
    public = models.BooleanField(default=False)
    date_added = models.DateTimeField(default=timezone.now)
    status = BoundedPositiveIntegerField(
        default=0,
        choices=(
            (ObjectStatus.ACTIVE, _("Active")),
            (ObjectStatus.PENDING_DELETION, _("Pending Deletion")),
            (ObjectStatus.DELETION_IN_PROGRESS, _("Deletion in Progress")),
        ),
        db_index=True,
    )
    first_event = models.DateTimeField(null=True)
    external_id = models.CharField(max_length=256, null=True)

    class flags(TypedClassBitField):
        has_releases: bool
        has_issue_alerts_targeting: bool
        has_transactions: bool
        has_sessions: bool
        has_profiles: bool
        has_replays: bool
        has_feedbacks: bool
        has_new_feedbacks: bool
        has_minified_stack_trace: bool
        has_cron_monitors: bool
        has_cron_checkins: bool
        # ...

class Release(Model):
    """
    A release is generally created when a new version is pushed into a
    production state.
    """
    organization = FlexibleForeignKey("sentry.Organization")
    projects = models.ManyToManyField(
        "sentry.Project", related_name="releases", through=ReleaseProject
    )
    status = BoundedPositiveIntegerField(
        default=ReleaseStatus.OPEN,
        choices=((ReleaseStatus.OPEN, _("Open")), (ReleaseStatus.ARCHIVED, _("Archived"))),
    )
    version = models.CharField(max_length=250)
    ref = models.CharField(max_length=250, null=True, blank=True)
    url = models.URLField(null=True, blank=True)
    date_added = models.DateTimeField(default=timezone.now, db_index=True)
    date_released = models.DateTimeField(null=True, blank=True)
    data = LegacyTextJSONField(default=dict)
    owner_id = HybridCloudForeignKey("sentry.User", on_delete="SET_NULL", null=True, blank=True)

    # 物化统计
    commit_count = BoundedPositiveIntegerField(null=True, default=0)
    last_commit_id = BoundedBigIntegerField(null=True)
    authors = ArrayField(models.TextField(), default=list, null=True)
    total_deploys = BoundedPositiveIntegerField(null=True, default=0)

    # 语义化版本反范式列
    package = models.TextField(null=True)
    major = models.BigIntegerField(null=True)
    minor = models.BigIntegerField(null=True)
    patch = models.BigIntegerField(null=True)
    revision = models.BigIntegerField(null=True)
    prerelease = models.TextField(null=True)
    build_code = models.TextField(null=True)
    build_number = models.BigIntegerField(null=True)

@cell_silo_model
class Group(Model):
    """
    Aggregated message which summarizes a set of Events.
    """
    project = FlexibleForeignKey("sentry.Project")
    logger = models.CharField(max_length=64, blank=True, default="", db_index=True)
    level = BoundedPositiveIntegerField(
        choices=[(key, str(val)) for key, val in sorted(LOG_LEVELS.items())],
        default=logging.ERROR, blank=True, db_index=True,
    )
    message = models.TextField()
    culprit = models.CharField(max_length=200, blank=True, null=True, db_column="view")
    num_comments = BoundedPositiveIntegerField(default=0, null=True)
    platform = models.CharField(max_length=64, null=True)
    status = BoundedPositiveIntegerField(default=GroupStatus.UNRESOLVED, db_index=True,
        choices=(
            (GroupStatus.UNRESOLVED, _("Unresolved")),
            (GroupStatus.RESOLVED, _("Resolved")),
            (GroupStatus.IGNORED, _("Ignored")),
        ))
    substatus = BoundedIntegerField(null=True,
        choices=(
            (GroupSubStatus.UNTIL_ESCALATING, _("Until escalating")),
            (GroupSubStatus.ONGOING, _("Ongoing")),
            (GroupSubStatus.ESCALATING, _("Escalating")),
            (GroupSubStatus.UNTIL_CONDITION_MET, _("Until condition met")),
            (GroupSubStatus.FOREVER, _("Forever")),
        ))
    times_seen = BoundedPositiveIntegerField(default=1, db_index=True)
    last_seen = models.DateTimeField(default=timezone.now, db_index=True)
    first_seen = models.DateTimeField(default=timezone.now, db_index=True)
    first_release = FlexibleForeignKey("sentry.Release", null=True, on_delete=models.PROTECT)
    resolved_at = models.DateTimeField(null=True, db_index=True)
    active_at = models.DateTimeField(null=True, db_index=True)
    time_spent_total = BoundedIntegerField(default=0)
    time_spent_count = BoundedIntegerField(default=0)
    data = LegacyTextJSONField(null=True)
    short_id = BoundedBigIntegerField(null=True)
    type = BoundedPositiveIntegerField(default=1, db_default=1, db_index=True)
    priority = models.PositiveIntegerField(null=True)
    priority_locked_at = models.DateTimeField(null=True)
    seer_fixability_score = models.FloatField(null=True)
    seer_autofix_last_triggered = models.DateTimeField(null=True)

class Meta:
    app_label = "sentry"
    db_table = "sentry_groupedmessage"
    verbose_name_plural = _("grouped messages")
    permissions = (("can_view", "Can view"),)
    indexes = [
        models.Index(fields=("project", "first_release")),
        models.Index(fields=("project", "id")),
        models.Index(fields=("project", "status", "last_seen", "id")),
        models.Index(fields=("project", "status", "type", "last_seen", "id")),
        models.Index(fields=("project", "status", "substatus", "last_seen", "id")),
        models.Index(fields=("project", "status", "substatus", "type", "last_seen", "id")),
        models.Index(fields=("project", "status", "substatus", "id")),
        models.Index(fields=("status", "substatus", "id")),
        models.Index(fields=("status", "substatus", "first_seen")),
        models.Index(fields=("project", "status", "priority", "last_seen", "id")),
    ]
    unique_together = (("project", "short_id"),)

def update_group_status(self, groups, status, substatus, activity_type, ...):
    selected_groups = Group.objects.filter(id__in=[g.id for g in groups]).exclude(
        status=status, substatus=substatus
    )
    for group in selected_groups:
        group.status = status
        group.substatus = substatus
        if status == GroupStatus.RESOLVED:
            group.resolved_at = timezone.now()

    Group.objects.bulk_update(modified_groups_list,
        ["status", "substatus", "priority", "resolved_at"])

    for group in modified_groups_list:
        Activity.objects.create_group_activity(group, activity_type, ...)
        record_group_history_from_activity_type(group, activity_type.value)

class GroupStatus:
    UNRESOLVED = 0
    RESOLVED = 1
    IGNORED = 2
    PENDING_DELETION = 3
    DELETION_IN_PROGRESS = 4
    PENDING_MERGE = 5
    REPROCESSING = 6
    # MUTED 是 IGNORED 的历史别名

def by_qualified_short_id_bulk(self, organization_id, short_ids_raw, *, project_ids):
    # 解析 PROJECT-32 格式
    # 首先按 project_slug 和 short_id 精确查询
    # 不命中时回退到 case-insensitive 的 slug 查询（处理历史遗留大小写问题）
    base_group_queryset = self.exclude(
        status__in=[GroupStatus.PENDING_DELETION, ...]
    ).filter(project__organization=organization_id)
    # ...

def from_event_id(self, project, event_id):
    event = eventstore.backend.get_event_by_id(project.id, event_id)
    if event:
        group_id = event.group_id
    if group_id is None:
        raise Group.DoesNotExist()
    return self.get(id=group_id)

def get_group_with_redirect(id_or_qualified_short_id, queryset=None, organization=None):
    try:
        return getter(**params), False
    except Group.DoesNotExist as error:
        # 查询 GroupRedirect 表
        params = {
            "id__in": GroupRedirect.objects.filter(
                previous_group_id=id_or_qualified_short_id,
            ).values_list("group_id", flat=True)[:1]
        }
        return updated_qs.get(**params), True  # True 表示发生了重定向

def bulk_get_latest_event_ids(groups):
    # 按 (organization_id, dataset) 分区
    for (organization_id, dataset), partition in partitions.items():
        request = Request(
            dataset=dataset.value,
            query=Query(
                match=Entity(dataset.value),
                select=[
                    Column("group_id"),
                    Function("argMax", [
                        Column("event_id"),
                        Function("tuple", [Column("timestamp"), Column("event_id")]),
                    ], "event_id"),
                ],
                groupby=[Column("group_id")],
                # ...
            ),
            tenant_ids={"organization_id": organization_id},
        )

class DjangoNodeStorage(NodeStorage):
    def _get_bytes(self, id: str) -> bytes | None:
        try:
            data = Node.objects.get(id=id).data
            return decompress(data)
        except Node.DoesNotExist:
            return None

    def _set_bytes(self, id, data, ttl=None):
        Node.objects.update_or_create(
            id=id, defaults={"data": compress(data), "timestamp": timezone.now()}
        )

    def cleanup(self, cutoff_timestamp):
        from sentry.db.deletion import BulkDeleteQuery
        total_seconds = (timezone.now() - cutoff_timestamp).total_seconds()
        days = math.floor(total_seconds / 86400)
        BulkDeleteQuery(model=Node, dtfield="timestamp", days=days).execute()

class Node(BaseModel):
    id = models.CharField(max_length=40, primary_key=True)
    data = models.TextField()  # 以 pickle 或压缩格式存储
    timestamp = models.DateTimeField(default=timezone.now, db_index=True)
    # db_table = "nodestore_node"

class BigtableNodeStorage(NodeStorage):
    store_class = BigtableKVStorage

    def __init__(self, project=None, instance="sentry", table="nodestore",
                 automatic_expiry=False, default_ttl=None,
                 compression=False, **client_options):
        self.store = self.store_class(
            project=project, instance=instance, table_name=table,
            default_ttl=default_ttl, compression=_compression,
        )
        self.skip_deletes = automatic_expiry and "_SENTRY_CLEANUP" in os.environ

    def _get_bytes(self, id: str) -> bytes | None:
        return self.store.get(id)

    def _set_bytes(self, id, data, ttl=None):
        self.store.set(id, data, ttl)

    def delete(self, id: str):
        if self.skip_deletes:
            return
        self.store.delete(id)

class NodeStorage(local, Service):
    @cached_property
    def cache(self) -> BaseCache | None:
        try:
            return caches["nodedata"]
        except InvalidCacheBackendError:
            return None

    def get(self, id, subkey=None):
        if subkey is None:
            item_from_cache = self._get_cache_item(id)
            if item_from_cache:
                return item_from_cache
        bytes_data = self._get_bytes(id)
        rv = self._decode(bytes_data, subkey=subkey)
        if subkey is None:
            self._set_cache_item(id, rv)  # 写入缓存
        return rv

    def _get_cache_item(self, item_id):
        if self.cache:
            return self.cache.get(item_id)
        return None

    def _set_cache_item(self, item_id, data):
        if self.cache and data:
            self.cache.set(item_id, data, timeout=options.get("nodestore.cache-ttl"))

def set_subkeys(self, item_id, data, ttl=None):
    """
    >>> nodestore.set_subkeys('key1', {
    ...    None: {'foo': 'bar'},
    ...    "reprocessing": {'foo': 'bam'},
    ... })
    """
    cache_item = data.get(None)
    bytes_data = self._encode(data)
    self.set_bytes(item_id, bytes_data, ttl=ttl)
    self._set_cache_item(item_id, cache_item)

def _encode(self, data):
    # data = {None: {...}, "reprocessing": {...}}
    lines = [json_dumps(data.pop(None)).encode("utf8")]
    for key, value in data.items():
        lines.append(key.encode("ascii"))
        lines.append(json_dumps(value).encode("utf8"))
    return b"\n".join(lines)

class Dataset(Enum):
    Events = "events"                  # 错误事件
    Transactions = "transactions"      # 性能事务
    Discover = "discover"             # Events + Transactions 联合查询
    Outcomes = "outcomes"             # 用量统计物化视图
    OutcomesRaw = "outcomes_raw"      # 原始用量统计
    Sessions = "sessions"             # 会话数据（已弃用）
    Metrics = "metrics"               # 发布健康度指标
    PerformanceMetrics = "generic_metrics"  # 通用性能指标
    Replays = "replays"               # 会话回放
    Profiles = "profiles"             # 性能剖析
    IssuePlatform = "search_issues"   # Issue Platform 搜索
    Functions = "functions"           # 函数级剖析
    SpansIndexed = "spans"            # Span 索引数据
    EventsAnalyticsPlatform = "events_analytics_platform"  # EAP

class EntityKey(Enum):
    Events = "events"
    Transactions = "transactions"
    Spans = "spans"
    EAPItemsSpan = "eap_items_span"
    EAPItems = "eap_items"
    IssuePlatform = "search_issues"
    Functions = "functions"
    MetricsSets = "metrics_sets"
    MetricsCounters = "metrics_counters"
    GenericMetricsDistributions = "generic_metrics_distributions"
    # ...

class Columns(Enum):
    EVENT_ID = Column(
        group_name="events.event_id",
        event_name="event_id",
        transaction_name="event_id",
        discover_name="event_id",
        issue_platform_name="event_id",
        alias="id",
    )
    GROUP_ID = Column(
        group_name="events.group_id",
        event_name="group_id",
        transaction_name=None,
        discover_name="group_id",
        issue_platform_name="group_id",
        alias="issue.id",
    )
    PROJECT_ID = Column(
        group_name="events.project_id",
        event_name="project_id",
        transaction_name="project_id",
        discover_name="project_id",
        issue_platform_name="project_id",
        alias="project.id",
    )

def get_recommended_event(group, conditions=None, start=None, end=None, verify_replay_exists=False):
    dataset = Dataset.Events if group.issue_category == GroupCategory.ERROR else Dataset.IssuePlatform

    events = eventstore.backend.get_events_snql(
        organization_id=group.project.organization_id,
        group_id=group.id,
        start=resolved_start,
        end=resolved_end,
        conditions=all_conditions,
        limit=limit,
        orderby=EventOrdering.RECOMMENDED.value,
        referrer="Group.get_helpful",
        dataset=dataset,
        tenant_ids={"organization_id": group.project.organization_id},
        inner_limit=1000,
    )

class EventOrdering(Enum):
    RECOMMENDED = [
        "-replay.id",
        "-trace.sampled",
        "num_processing_errors",
        "-profile.id",
        "-timestamp",
        "-id",
    ]

def query(
    selected_columns, query, snuba_params, ...,
    dataset: Dataset = Dataset.Discover,
    fallback_to_transactions: bool = False,
    referrer: str,
) -> EventsResponse:
    return discover.query(
        selected_columns, query, snuba_params=snuba_params,
        dataset=dataset, referrer=referrer, ...
    )

def query(
    selected_columns, query, snuba_params,
    equations=None, orderby=None, offset=None, limit=50,
    auto_fields=False, auto_aggregations=False,
    use_aggregate_conditions=False, conditions=None,
    functions_acl=None, transform_alias_to_input_format=False,
    sample=None, has_metrics=False, skip_tag_resolution=False,
    dataset=Dataset.Discover,
    referrer=None,
) -> EventsResponse:

class SnubaQuery(Model):
    class Type(Enum):
        ERROR = 0
        PERFORMANCE = 1
        CRASH_RATE = 2

    environment = FlexibleForeignKey("sentry.Environment", null=True, db_constraint=False)
    type = models.SmallIntegerField()
    dataset = models.TextField()
    query = models.TextField()
    group_by = ArrayField(models.CharField(max_length=200), null=True, size=100)
    aggregate = models.TextField()
    time_window = models.IntegerField()   # 聚合窗口（秒）
    resolution = models.IntegerField()    # 评估间隔（秒）
    extrapolation_mode = models.IntegerField()
    query_snapshot = models.JSONField(null=True)
    date_added = models.DateTimeField(default=timezone.now)

class QuerySubscription(Model):
    class Status(Enum):
        ACTIVE = 0
        CREATING = 1
        UPDATING = 2
        DELETING = 3
        DISABLED = 4

    project = FlexibleForeignKey("sentry.Project", db_constraint=False)
    snuba_query = FlexibleForeignKey("sentry.SnubaQuery", related_name="subscriptions")
    type = models.TextField()
    status = models.SmallIntegerField(default=Status.ACTIVE.value, db_index=True)
    subscription_id = models.TextField(unique=True, null=True)
    date_added = models.DateTimeField(default=timezone.now)
    query_extra = models.TextField(null=True)

class GroupManager(BaseManager["Group"]):
    # ...
# Group 模型声明
objects: ClassVar[GroupManager] = GroupManager(cache_fields=("id",))

class QuerySubscription(Model):
    objects: ClassVar[BaseManager[Self]] = BaseManager(
        cache_fields=("pk", "subscription_id"),
        cache_ttl=int(timedelta(hours=1).total_seconds())
    )

extra_data = (
    {
        "group_id": event.group_id,
        "group_ids": [group.id for group in getattr(event, "groups", [])],
        "event_id": event.event_id,
        "organization_id": project.organization_id,
        "project_id": event.project_id,
        "message": event.search_message,
        "platform": event.platform,
        "datetime": json.datetime_to_str(event.datetime),
        "data": event_data,
        "primary_hash": primary_hash,
        "retention_days": retention_days,
        "occurrence_id": occurrence_data.get("id"),
        "occurrence_data": occurrence_data,
    },
    {
        "is_new": is_new,
        "is_regression": is_regression,
        "is_new_group_environment": is_new_group_environment,
        "skip_consume": skip_consume,
        "group_states": group_states,
    },
)

def get_task_kwargs_for_insert(operation, event_data, task_state):
    if task_state and task_state.get("skip_consume", False):
        return None
    kwargs = {
        "event_id": event_data["event_id"],
        "project_id": event_data["project_id"],
        "group_id": event_data["group_id"],
        "primary_hash": event_data["primary_hash"],
        "occurrence_id": event_data.get("occurrence_id"),
    }
    for name in ("is_new", "is_regression", "is_new_group_environment"):
        kwargs[name] = task_state[name]
    return kwargs

class KafkaEventStream(SnubaProtocolEventStream):
    def __init__(self, **options):
        super().__init__(**options)
        self.topic = Topic.EVENTS
        self.transactions_topic = Topic.TRANSACTIONS
        self.issue_platform_topic = Topic.EVENTSTREAM_GENERIC

    def _send(self, project_id, _type, extra_data=(), asynchronous=True,
              headers=None, skip_semantic_partitioning=False,
              event_type=EventStreamEventType.Error):
        if event_type == EventStreamEventType.Transaction:
            topic = self.get_transactions_topic(project_id)
        elif event_type == EventStreamEventType.Generic:
            topic = self.issue_platform_topic
        else:
            topic = self.topic

        producer = self.get_producer(topic)
        real_topic = get_topic_definition(topic)["real_topic_name"]

        produce_future = producer.produce(
            destination=ArroyoTopic(real_topic),
            payload=KafkaPayload(
                key=str(project_id).encode("utf-8") if not skip_semantic_partitioning else None,
                value=json.dumps((self.EVENT_PROTOCOL_VERSION, _type) + extra_data).encode("utf-8"),
                headers=[(k, v.encode("utf-8")) for k, v in headers.items()],
            ),
        )

# Nodestore Django 后端的清理
def cleanup(self, cutoff_timestamp):
    from sentry.db.deletion import BulkDeleteQuery
    total_seconds = (timezone.now() - cutoff_timestamp).total_seconds()
    days = math.floor(total_seconds / 86400)
    BulkDeleteQuery(model=Node, dtfield="timestamp", days=days).execute()

expired, start = outside_retention_with_modified_start(
    start, end, Organization(organization_id)
)
if expired:
    return None  # 数据已过期，跳过查询