# factory.py:78-97
class IngestStrategyFactory(ProcessingStrategyFactory[KafkaPayload]):
    def __init__(
        self,
        consumer_type: str,
        reprocess_only_stuck_events: bool,
        reprocess_only_events_not_in_nodestore: bool,
        stop_at_timestamp: int | None,
        num_processes: int,
        max_batch_size: int,
        max_batch_time: int,
        input_block_size: int | None,
        output_block_size: int | None,
    ):

# types.py:4-12
class ConsumerType(StrEnum):
    Events = "events"          # 普通错误事件 (ingest-events)
    Attachments = "attachments" # 附件事件 (ingest-attachments)
    Transactions = "transactions" # 事务事件 (ingest-transactions)
    Feedback = "feedback"       # 用户反馈 (ingest-feedback-events)

# simple_event.py:54-55
message: IngestMessage = msgpack.unpackb(raw_payload, use_list=False)
# 消息结构:
# {
#     "type": "event",
#     "project_id": 123,
#     "event_id": "abc...",
#     "payload": b'{"event_id":"abc...","project":123,...}',  # JSON 字节
#     "start_time": 1691234567.890,
#     "remote_addr": "1.2.3.4",
#     "attachments": [...]  # 可选
# }

# attachment_event.py:47-55
if message["type"] == "attachment_chunk":
    # 附件分片，先于 event 处理
    if not reprocess_only_stuck_events:
        process_attachment_chunk(message)
    return None  # 已消费，不进入第二步
# 返回其他类型消息进入第二步处理

# factory.py:118-190
def create_with_partitions(self, commit, partitions):
    final_step = CommitOffsets(commit)  # 提交 Kafka offset
    
    if not self.is_attachment_topic:
        # 普通 Topic: 单阶段流水线
        event_function = partial(
            process_simple_event_message,
            consumer_type=self.consumer_type, ...
        )
        next_step = maybe_multiprocess_step(mp, event_function, final_step, self._pool)
        return maybe_backpressure_step(
            health_checker=self.health_checker,
            next_step=next_step,
            is_recovery=self.reprocess_only_stuck_events,
        )
    
    # Attachments Topic: 两阶段流水线
    # Step 1: 处理 attachment_chunk（必须先完成）
    # Step 2: 处理 event / attachment / user_report
    step_2 = maybe_multiprocess_step(mp, processing_function, final_step, self._attachments_pool)
    filter_step = FilterStep(function=filter_fn, next_step=step_2)
    step_1 = maybe_multiprocess_step(mp, attachment_function, filter_step, self._pool)
    return maybe_backpressure_step(...)

# processors.py:117-131
with start_span(op="deduplication_check", name="deduplication_check"):
    deduplication_key = f"ev:{project_id}:{event_id}"
    cached_value = cache.get(deduplication_key)
    if cached_value is not None:
        logger.warning(
            "pre-process-forwarder detected a duplicated event with id:%s for project:%s.",
            event_id, project_id,
        )
        return  # 已处理过，跳过

# processors.py:309-310
cache.set(deduplication_key, "", CACHE_TIMEOUT)  # CACHE_TIMEOUT=3600

# processors.py:133-143
if killswitch_matches_context(
    "store.load-shed-pipeline-projects",
    {
        "project_id": project_id,
        "event_id": event_id,
        "has_attachments": bool(attachments),
    },
):
    return  # 静默丢弃

# event_manager.py:343-384
class EventManager:
    def __init__(
        self,
        data: MutableMapping[str, Any],
        version: str = "5",
        project: Project | None = None,
        grouping_config: GroupingConfig | None = None,
        client_ip: str | None = None,
        user_agent: str | None = None,
        auth: Any | None = None,
        key: Any | None = None,
        content_encoding: str | None = None,
        is_renormalize: bool = False,
        remove_other: bool | None = None,
        project_config: Any | None = None,
        sent_at: datetime | None = None,
    ):

# event_manager.py:386-425
def normalize(self, project_id: int | None = None) -> None:
    with metrics.timer("events.store.normalize.duration"):
        self._normalize_impl(project_id=project_id)

def _normalize_impl(self, project_id: int | None = None) -> None:
    # ...
    from sentry_relay.processing import StoreNormalizer

    rust_normalizer = StoreNormalizer(
        project_id=self._project.id if self._project else project_id,
        client_ip=self._client_ip,
        client=self._auth.client if self._auth else None,
        key_id=str(self._key.id) if self._key else None,
        grouping_config=self._grouping_config,
        protocol_version=str(self.version) if self.version is not None else None,
        is_renormalize=self._is_renormalize,
        remove_other=self._remove_other,
        normalize_user_agent=True,
        sent_at=self.sent_at.isoformat() if self.sent_at is not None else None,
        json_dumps=orjson.dumps,
        **DEFAULT_STORE_NORMALIZER_ARGS,
    )

    self._data = rust_normalizer.normalize_event(dict(self._data), json_loads=orjson.loads)

# event_manager.py:421-425
if pre_normalize_type in ("generic", "feedback"):
    self._data["type"] = pre_normalize_type

# event_manager.py:431-508
def save(
    self,
    project_id: int | None = None,
    project: Project | None = None,
    raw: bool = False,
    assume_normalized: bool = False,
    start_time: float | None = None,
    cache_key: str | None = None,
    skip_send_first_transaction: bool = False,
    attachments: list[CachedAttachment] | None = None,
) -> Event:

    event_type = self._data.get("type")
    if event_type == "transaction":
        # 事务事件：走 save_transaction_events 路径
        jobs = save_transaction_events([job], projects, skip_send_first_transaction)
        return jobs[0]["event"]
    elif event_type == "generic":
        # 通用事件（Issue Platform）：走 save_generic_events 路径
        jobs = save_generic_events([job], projects)
        return jobs[0]["event"]
    else:
        # 错误事件：走 save_error_events 路径
        return self.save_error_events(project, job, projects, metric_tags, attachments or [], raw, cache_key)

# event_manager.py:1076-1104
def _nodestore_save_many(jobs: Sequence[Job], app_feature: str) -> None:
    inserted_time = datetime.now(timezone.utc).timestamp()
    for job in jobs:
        subkeys = {}
        event = job["event"]
        # 错误事件：保留未处理的事件数据作为 subkey
        if event.get_event_type() not in ("transaction", "generic") and job["groups"]:
            unprocessed = event_processing_store.get(
                cache_key_for_event({"project": event.project_id, "event_id": event.event_id}),
                unprocessed=True,
            )
            if unprocessed is not None:
                subkeys["unprocessed"] = unprocessed

        job["event"].data["nodestore_insert"] = inserted_time
        job["event"].data.save(subkeys=subkeys)

# base.py:39-195
class EventStream(Service):
    __all__ = (
        "insert",           # 插入事件
        "start_delete_groups", "end_delete_groups",  # 删除分组
        "start_merge", "end_merge",                   # 合并分组
        "start_unmerge", "end_unmerge",               # 拆分分组
        "start_delete_tag", "end_delete_tag",         # 删除标签
        "tombstone_events_unsafe",                     # 标记事件删除
        "replace_group_unsafe",                       # 替换分组
        "exclude_groups",                              # 排除分组
        "requires_post_process_forwarder",
    )

# kafka/backend.py:34-37
class KafkaEventStream(SnubaProtocolEventStream):
    def __init__(self, **options: Any) -> None:
        self.topic = Topic.EVENTS                # 错误事件
        self.transactions_topic = Topic.TRANSACTIONS  # 事务事件
        self.issue_platform_topic = Topic.EVENTSTREAM_GENERIC  # 通用事件

# base.py:60-92
def _dispatch_post_process_group_task(self, event_id, project_id, group_id, ...):
    cache_key = cache_key_for_event({"project": project_id, "event_id": event_id})
    post_process_group.apply_async(
        kwargs={
            "is_new": is_new,
            "is_regression": is_regression,
            "is_new_group_environment": is_new_group_environment,
            "primary_hash": primary_hash,
            "cache_key": cache_key,
            "group_id": group_id,
            "group_states": group_states,
            "occurrence_id": occurrence_id,
            "project_id": project_id,
            "eventstream_type": eventstream_type,
        },
        headers={"sentry-propagate-traces": False},
    )

class GroupingConfig(TypedDict):
    id: str              # 分组策略ID，如 "newstyle:2023-01-11"
    enhancements: str    # Base64编码的增强规则

class PrimaryGroupingConfigLoader(ProjectGroupingConfigLoader):
    option_name = "sentry:grouping_config"
    cache_prefix = "grouping-enhancements:"

# variants.py:26-77
class BaseVariant(ABC):
    variant_name: str | None = None
    @property
    def contributes(self) -> bool: return True
    @property
    @abstractmethod
    def type(self) -> str: ...
    def get_hash(self) -> str | None: return None
    @property
    def key(self) -> str: return self.type
    @property
    def description(self) -> str: return self.type
    @property
    def hint(self) -> str | None: return None

event.data["fingerprint"] = event.data.data.get("fingerprint") or [""]
apply_server_side_fingerprinting(
    event.data.data, get_fingerprinting_config_for_project(project)
)

def _calculate_event_grouping(
    project: Project, event: Event, grouping_config: GroupingConfig
) -> tuple[list[str], dict[str, BaseVariant]]:
    loaded_grouping_config = load_grouping_config(grouping_config)

    # 1. 应用服务端指纹规则
    event.data["fingerprint"] = event.data.data.get("fingerprint") or [""]
    apply_server_side_fingerprinting(
        event.data.data, get_fingerprinting_config_for_project(project)
    )

    # 2. 标准化堆栈跟踪用于分组
    event.normalize_stacktraces_for_grouping(loaded_grouping_config)

    # 3. 计算哈希和变体
    hashes, variants = event.get_hashes_and_variants(loaded_grouping_config)

    return (hashes, variants)

# 主分组
primary = get_hashes_and_grouphashes(job, run_primary_grouping, metric_tags)
if primary.existing_grouphash:
    group_info = handle_existing_grouphash(...)
else:
    # 辅助分组（过渡期使用）
    secondary = get_hashes_and_grouphashes(job, maybe_run_secondary_grouping, metric_tags)
    all_grouphashes = primary.grouphashes + secondary.grouphashes
    if secondary.existing_grouphash:
        group_info = handle_existing_grouphash(...)
    else:
        # Seer ML 匹配 or 创建新 Group

# event_manager.py:1344-1353
seer_matched_grouphash = maybe_check_seer_for_matching_grouphash(
    event, primary.grouphashes[0], primary.variants, all_grouphashes
)
if seer_matched_grouphash:
    group_info = handle_existing_grouphash(job, seer_matched_grouphash, all_grouphashes)
else:
    group_info = create_group_with_grouphashes(job, all_grouphashes)

with transaction.atomic(router.db_for_write(GroupHash)):
    # 获取锁，防止并发创建
    group = Group.objects.create(
        project=project,
        short_id=project.next_short_id(),
        **kwargs  # 包含 first_seen, data, platform, message, level, culprit 等
    )
    add_group_id_to_grouphashes(group, grouphashes)

# post_process.py 中的核心数据结构
class PostProcessJob(TypedDict, total=False):
    event: GroupEvent
    group_state: GroupState
    is_reprocessed: bool
    has_reappeared: bool
    has_escalated: bool
    halt_post_process: bool

# post_process.py:56-60
locks = LockManager(
    build_instance_from_options_of_type(
        LockBackend, settings.SENTRY_POST_PROCESS_LOCKS_BACKEND_OPTIONS
    )
)

# inbound_filters.py:37-49
FILTER_STAT_KEYS_TO_VALUES = {
    FilterStatKeys.IP_ADDRESS: TSDBModel.project_total_received_ip_address,
    FilterStatKeys.RELEASE_VERSION: TSDBModel.project_total_received_release_version,
    FilterStatKeys.ERROR_MESSAGE: TSDBModel.project_total_received_error_message,
    FilterStatKeys.BROWSER_EXTENSION: TSDBModel.project_total_received_browser_extensions,
    FilterStatKeys.LEGACY_BROWSER: TSDBModel.project_total_received_legacy_browsers,
    FilterStatKeys.LOCALHOST: TSDBModel.project_total_received_localhost,
    FilterStatKeys.WEB_CRAWLER: TSDBModel.project_total_received_web_crawlers,
    # ...
}

ACTIVE_GENERIC_FILTERS: Sequence[tuple[str, Callable[[], RuleCondition | None]]] = [
    ("chunk-load-error", _chunk_load_error_filter),
    ("react-hydration-errors", _hydration_error_filter),
    ("custom-error", _custom_error_filter),
]

values = [
    ("ChunkLoadError", "Loading chunk *"),
    ("*Uncaught *", "ChunkLoadError: Loading chunk *"),
    ("ChunkLoadError", "Failed to load chunk *"),
    # ...
]

values = [
    (None, "*https://reactjs.org/docs/error-decoder.html?invariant={418,419,421,422,423,425}*"),
    (None, "*https://react.dev/errors/{418,419,421,422,423,425}*"),
]

# inbound_filters.py:83-119
def set_filter_state(filter_id, project: Project, state):
    # 旧浏览器过滤器特殊处理：支持 active / subfilters
    if flt == _legacy_browsers_filter:
        # option_val 可能是 "0", "1", 或 sorted subfilters 列表
        ProjectOption.objects.set_value(
            project=project, key=f"filters:{filter_id}", value=option_val
        )
    else:
        # 所有布尔过滤器
        ProjectOption.objects.set_value(
            project=project,
            key=f"filters:{filter_id}",
            value="1" if state.get("active", False) else "0",
        )

# quotas/base.py:27-34
@unique
class QuotaScope(IntEnum):
    ORGANIZATION = 1   # 组织级别
    PROJECT = 2        # 项目级别
    KEY = 3            # DSN Key 级别

@dataclass
class AbuseQuota:
    id: str
    option: str
    categories: list[DataCategory]
    scope: AbuseQuotaScope
    namespace: str | None = None

def get_all_pii_configs(project):
    # 1. 组织和项目级别的自定义 PII 规则
    pii_config = get_pii_config(project)
    if pii_config:
        yield pii_config

    # 2. 数据脱敏设置（敏感字段、IP 脱敏等）
    settings = get_datascrubbing_settings(project)
    yield convert_datascrubbing_config(settings, json_dumps=orjson.dumps, json_loads=orjson.loads)

# datascrubbing.py:112-144
def _merge_pii_configs(prefixes_and_configs):
    # 为每个配置源添加前缀，确保规则名唯一
    # 组织规则前缀: "organization:"
    # 项目规则前缀: "project:"
    for prefix, partial_config in prefixes_and_configs:
        for rule_name, rule in rules.items():
            prefixed_rule_name = f"{prefix}{rule_name}"
            merged_config.setdefault("rules", {})[prefixed_rule_name] = (...)

def scrub_data(project: Project, event: MutableMapping[str, Any]) -> MutableMapping[str, Any]:
    for config in get_all_pii_configs(project):
        event = pii_strip_event(
            config, dict(event), json_loads=orjson.loads, json_dumps=orjson.dumps
        )
    return event

# process_event 中的第二轮脱敏 (store.py:370-376):
if has_changed:
    new_data = safe_execute(scrub_data, project=project, event=data)
    if new_data is not None:
        data = new_data

def get_datascrubbing_settings(project):
    rv = {}
    # 安全字段（不被脱敏的字段）
    rv["excludeFields"] = org.get_option("sentry:safe_fields", []) + \
                           project.get_option("sentry:safe_fields", [])
    # 数据脱敏开关
    if org.get_option("sentry:require_scrub_data", False) or \
       project.get_option("sentry:scrub_data", True):
        rv["scrubData"] = True
    # IP 脱敏开关
    if org.get_option("sentry:require_scrub_ip_address", False) or \
       project.get_option("sentry:scrub_ip_address", False):
        rv["scrubIpAddresses"] = True
    # 敏感字段列表
    rv["sensitiveFields"] = org.get_option("sentry:sensitive_fields", []) + \
                             project.get_option("sentry:sensitive_fields", [])
    return rv

class MultiProcessConfig(NamedTuple):
    num_processes: int          # 进程数
    max_batch_size: int         # 最大批量大小
    max_batch_time: int         # 最大批处理时间（毫秒）
    input_block_size: int | None  # 输入分块大小
    output_block_size: int | None # 输出分块大小

# factory.py:111-114
if num_processes > 1:
    self.multi_process = MultiProcessConfig(
        num_processes, max_batch_size, max_batch_time, input_block_size, output_block_size
    )

def maybe_multiprocess_step(mp, function, next_step, pool):
    if mp is not None:
        return run_task_with_multiprocessing(
            function=function, next_step=next_step,
            max_batch_size=mp.max_batch_size,
            max_batch_time=mp.max_batch_time,
            pool=pool, ...
        )
    else:
        return RunTask(function=function, next_step=next_step)

def maybe_backpressure_step(health_checker, next_step, is_recovery):
    if is_recovery:
        return next_step  # 恢复消费者不加背压（只处理卡住的事件）
    else:
        return create_backpressure_step(health_checker=health_checker, next_step=next_step)

# processors.py:154-157
if consumer_type == ConsumerType.Transactions or data.get("type") == "transaction":
    processing_store = transaction_processing_store
else:
    processing_store = event_processing_store

# store.py:98-119
def submit_save_event(task_kind, project_id, cache_key, event_id, start_time, data, inline=False):
    if cache_key:
        data = None  # 不传递数据，由任务自行从Redis读取
    
    task_kwargs = {
        "cache_key": cache_key,
        "data": data,
        "start_time": start_time,
        "event_id": event_id,
        "project_id": project_id,
    }

# store.py:599-605 (transactions 立即清理)
if consumer_type == ConsumerType.Transactions and event_id:
    if cache_key:
        processing_store.delete_by_key(cache_key)