class IntegrationDomain(StrEnum):
    MESSAGING = "messaging"           # 即时通讯：Slack、Discord、MS Teams
    PROJECT_MANAGEMENT = "project_management"  # 项目管理：Jira、Jira Server
    SOURCE_CODE_MANAGEMENT = "source_code_management"  # 源码管理：GitHub、GitLab、Bitbucket、Azure DevOps
    ON_CALL_SCHEDULING = "on_call_scheduling"  # 值班调度：PagerDuty、OpsGenie
    IDENTITY = "identity"             # 身份认证管道
    GENERAL = "general"               # 跨领域通用

class IntegrationFeatures(StrEnum):
    ALERT_RULE = "alert-rule"                # 告警规则通知
    CHAT_UNFURL = "chat-unfurl"              # 聊天链接展开
    COMMITS = "commits"                      # 提交数据关联
    ISSUE_BASIC = "issue-basic"              # 基本问题链接
    ISSUE_SYNC = "issue-sync"                # 问题双向同步
    STACKTRACE_LINK = "stacktrace-link"      # 堆栈跟踪链接
    CODEOWNERS = "codeowners"                # CODEOWNERS 文件
    TICKET_RULES = "ticket-rules"            # 自动创建工单
    SERVERLESS = "serverless"                # 无服务器部署
    MOBILE = "mobile"                        # 移动端
    DATA_FORWARDING = "data-forwarding"     # 数据转发（Plugin 专用）
    DEPLOYMENT = "deployment"               # 部署集成
    CODING_AGENT = "coding-agent"           # 编码代理
    MONITORING = "monitoring"               # 监控

class IntegrationManager:
    def __init__(self):
        self.__values: dict[str, type[IntegrationProvider]] = {}

    def register(self, cls: type[IntegrationProvider]) -> None:
        self.__values[cls.key] = cls

    def get(self, key: str) -> IntegrationProvider:
        try:
            cls = self.__values[key]
        except KeyError:
            raise NotRegistered(key)
        return cls()

    def all(self) -> Iterable[IntegrationProvider]:
        for key in self.__values.keys():
            integration = self.get(key)
            if integration.visible:
                yield integration

# 全局单例
default_manager = IntegrationManager()
all = default_manager.all
get = default_manager.get
register = default_manager.register

class IntegrationMetadata(NamedTuple):
    description: str       # Markdown 格式的描述文本
    features: Sequence[FeatureDescription]  # 功能列表
    author: str            # 作者名称
    noun: str              # 安装单元的称谓（如 "Installation"、"Workspace"）
    issue_url: str         # 问题反馈链接
    source_url: str        # 源代码链接
    aspects: dict[str, Any]  # 扩展配置（如外部安装入口）

class FeatureDescription(NamedTuple):
    description: str                        # 功能的 Markdown 描述
    featureGate: IntegrationFeatures        # 对应的功能开关

FEATURES = [
    FeatureDescription(
        """
        Unfurls Sentry URLs directly within Slack, providing you context and
        actionability on issues right at your fingertips.
        """,
        IntegrationFeatures.CHAT_UNFURL,
    ),
    FeatureDescription(
        """
        Configure rule based Slack notifications to automatically be posted into a
        specific channel.
        """,
        IntegrationFeatures.ALERT_RULE,
    ),
]

metadata = IntegrationMetadata(
    description=DESCRIPTION.strip(),
    features=FEATURES,
    author="The Sentry Team",
    noun=_("Workspace"),
    issue_url="https://github.com/getsentry/sentry/issues/new?...",
    source_url="https://github.com/getsentry/sentry/tree/master/src/sentry/integrations/slack",
    aspects={},
)

def finish_pipeline(self) -> HttpResponseBase:
    # 1. 权限检查：用户必须是 org owner/manager/admin
    # 2. 调用 provider.build_integration(state) 构建 IntegrationData
    # 3. 调用 _finish_pipeline(data) 完成安装
    #    a. 创建/更新 Integration 数据库记录
    #    b. 关联 Identity（如果有 user_identity）
    #    c. 创建 OrganizationIntegration 关联记录
    # 4. 创建审计日志
    # 5. 调用 provider.post_install()
    # 6. 返回 dialog_success 响应

def _install_integration(self, data: IntegrationData) -> OrganizationIntegration:
    # 如果 data 包含 expect_exists，直接查找已有 Integration
    if "expect_exists" in data:
        self.integration = Integration.objects.get(
            provider=self.provider.integration_key,
            external_id=data["external_id"],
        )
    else:
        # 否则调用 ensure_integration 创建或更新
        self.integration = ensure_integration(self.provider.integration_key, data)

    # 处理用户身份 Identity
    identity = data.get("user_identity")
    if identity:
        idp = IdentityProvider.objects.get_or_create(
            external_id=data.get("idp_external_id", data["external_id"]),
            type=identity["type"],
        )
        identity_model = Identity.objects.link_identity(
            user=self.request.user,
            idp=idp,
            external_id=identity["external_id"],
        )

    # 创建 OrganizationIntegration
    org_integration = self.integration.add_organization(
        self.organization, self.request.user,
        default_auth_id=identity_model.id if self.provider.needs_default_identity else None,
    )
    return org_integration

class ApiPipelineStep:
    step_name: str  # 步骤标识符

    def get_step_data(self, pipeline, request) -> dict:
        """GET 请求时返回初始数据"""

    def get_serializer_cls(self) -> type | None:
        """返回验证 POST 数据的序列化器类"""

    def handle_post(self, validated_data, pipeline, request) -> PipelineStepResult:
        """处理 POST 请求，返回 advance/error/stay/complete"""

class OAuthLoginApiStep:
    step_name = "oauth_login"

    def get_step_data(self, pipeline, request):
        return {"oauthUrl": build_github_oauth_url(pipeline)}

    def handle_post(self, validated_data, pipeline, request):
        # 1. 验证 state 签名
        if validated_data["state"] != pipeline.signature:
            return PipelineStepResult.error("Invalid state")
        # 2. 用 code 换取 access_token
        result = exchange_github_oauth(code=validated_data["code"])
        # 3. 绑定状态供下一步使用
        pipeline.bind_state("github_authenticated_user", result.authenticated_user)
        pipeline.bind_state("existing_installation_info", result.installation_info)
        return PipelineStepResult.advance()

class GithubOrganizationSelectionApiStep:
    step_name = "org_selection"

    def get_step_data(self, pipeline, request):
        installations = pipeline.fetch_state("existing_installation_info")
        return {
            "installAppUrl": get_install_app_url(),
            "installationInfo": [
                {"installationId": i["installation_id"], "githubAccount": i["github_account"]}
                for i in (installations or [])
            ],
        }

    def handle_post(self, validated_data, pipeline, request):
        # 用户选择了某个已有安装
        if chosen_id := validated_data.get("chosen_installation_id"):
            validate_org_installation_choice(chosen_id, pipeline)
            pipeline.bind_state("installation_id", chosen_id)
        # 用户完成了 GitHub App 安装弹窗
        if post_id := validated_data.get("installation_id"):
            pipeline.bind_state("installation_id", post_id)
        # 验证 installation 有效性
        installation_id = validate_github_installation(pipeline)
        if installation_id is None:
            return PipelineStepResult.stay(data={"installAppUrl": get_install_app_url()})
        return PipelineStepResult.advance()

class IntegrationData(TypedDict):
    external_id: str         # 必填：第三方服务的唯一标识
    name: NotRequired[str]   # 显示名称
    metadata: NotRequired[dict[str, Any]]  # 集成级元数据（API token 等）
    post_install_data: NotRequired[dict[str, Any]]  # 传给 post_install 的数据
    expect_exists: NotRequired[bool]  # 如果为 True，只查找不创建
    user_identity: NotRequired[_UserIdentity]  # 安装者的身份信息

def build_integration(self, state: Mapping[str, Any]) -> IntegrationData:
    data = state["oauth_data"]
    return {
        "name": data["team"]["name"],
        "external_id": data["team"]["id"],
        "metadata": {
            "access_token": data["access_token"],
            "scopes": sorted(scopes),
            "icon": team_data["icon"]["image_132"],
            "domain_name": team_data["domain"] + ".slack.com",
            "installation_type": "born_as_bot",
        },
        "user_identity": {
            "type": "slack",
            "external_id": data["authed_user"]["id"],
            "scopes": [],
            "data": {},
        },
    }

# GitHub 示例 (github/integration.py:718-746)
def post_install(self, integration, organization, *, extra):
    # 异步迁移已有的 repos
    repos = repository_service.get_repositories(
        organization_id=organization.id, providers=[self.key], has_integration=False
    )
    for repo in repos:
        migrate_repo.apply_async(kwargs={
            "repo_id": repo.id, "integration_id": integration.id,
            "organization_id": organization.id,
        })
    # 异步链接所有可访问的 repos
    link_all_repos.apply_async(kwargs={
        "integration_key": self.key, "integration_id": integration.id,
        "organization_id": organization.id,
    })

class IntegrationInstallation(abc.ABC):
    def __init__(self, model: RpcIntegration | Integration, organization_id: int) -> None:
        self.model = model          # Integration 模型实例
        self.organization_id = organization_id  # 组织 ID

    @cached_property
    def org_integration(self) -> RpcOrganizationIntegration:
        """缓存的组织-集成关联记录"""
        return integration_service.get_organization_integration(
            integration_id=self.model.id, organization_id=self.organization_id,
        )

    @cached_property
    def organization(self) -> RpcOrganization:
        """缓存的 RpcOrganization"""
        return organization_service.get(id=self.organization_id)

@classmethod
def get_installation(cls, model, organization_id, **kwargs):
    return cls.integration_cls(model, organization_id, **kwargs)

# GitHub - 使用 installation token
class GitHubIntegration(RepositoryIntegration, ...):
    def get_client(self) -> GitHubBaseClient:
        return GitHubApiClient(
            integration=self.model,
            org_integration_id=self.org_integration.id,
        )

# Slack - 使用 SlackSdkClient
class SlackIntegration(IntegrationInstallation):
    def get_client(self) -> SlackSdkClient:
        return SlackSdkClient(integration_id=self.model.id)

# Jira - 使用 OAuth 签名
class JiraIntegration(IssueSyncIntegration):
    def get_client(self) -> JiraCloudClient:
        return JiraCloudClient(
            integration=self.model,
            verify_ssl=True,
        )

def get_organization_config(self) -> list[dict[str, Any]]:
    config = [
        {
            "name": self.outbound_status_key,
            "type": "choice_mapper",
            "label": "Sync Sentry Status to GitHub",
            "help": "When a Sentry issue changes status...",
            "addButtonText": "Add GitHub Project",
            "addDropdown": {
                "emptyMessage": "All projects configured",
                "items": current_repo_items,  # 已配置的项目
                "url": reverse("sentry-integration-github-search", ...),
            },
            "mappedSelectors": {
                "on_resolve": {"choices": [("open", "Open"), ("closed", "Closed")]},
                "on_unresolve": {"choices": [("open", "Open"), ("closed", "Closed")]},
            },
        },
        {"name": self.comment_key, "type": "boolean", "label": "Sync Sentry Comments to GitHub"},
        {"name": "pr_comments", "type": "boolean", "label": "Enable Comments on Suspect Pull Requests"},
    ]
    return config

def update_organization_config(self, data):
    config = self.org_integration.config

    if "sync_status_forward" in data:
        project_mappings = data.pop("sync_status_forward")
        # 验证必填字段
        if any(not m["on_unresolve"] or not m["on_resolve"]
               for m in project_mappings.values()):
            raise IntegrationError("Resolve and unresolve status are required.")

        # 清空并重建 IntegrationExternalProject 记录
        IntegrationExternalProject.objects.filter(
            organization_integration_id=self.org_integration.id
        ).delete()
        for project_id, statuses in project_mappings.items():
            IntegrationExternalProject.objects.create(
                organization_integration_id=self.org_integration.id,
                external_id=project_id,
                resolved_status=statuses["on_resolve"],
                unresolved_status=statuses["on_unresolve"],
            )

    # 合并并持久化
    config.update(data)
    org_integration = integration_service.update_organization_integration(
        org_integration_id=self.org_integration.id, config=config,
    )
    self.org_integration = org_integration

# Vercel 集成卸载时调用 Vercel API 移除 webhook
def uninstall(self):
    client = self.get_client()
    try:
        client.uninstall(self.get_configuration_id())
    except ApiError as error:
        if error.code == 403:
            pass  # 已卸载，忽略
        else:
            raise

# GitHub
def _get_debug_metadata_keys(self) -> list[str]:
    return ["account_type", "domain_name", "permissions"]

# Slack
def _get_debug_metadata_keys(self) -> list[str]:
    return ["domain_name", "installation_type", "scopes"]

class IssueBasicIntegration(IntegrationInstallation, ABC):
    @abstractmethod
    def get_issue_url(self, key: str) -> str:
        """给定外部 issue key，返回外部 issue 链接"""

    def get_group_body(self, group, event, **kwargs):
        """构建 issue 描述文本"""

    def get_group_title(self, group, event, **kwargs):
        """构建 issue 标题"""

    # 创建和链接 issue 的配置
    def get_create_issue_config(self, group, user, **kwargs):
        """返回创建 issue 时的表单配置"""

    def get_link_issue_config(self, group, **kwargs):
        """返回链接已有 issue 时的表单配置"""

    @abstractmethod
    def create_issue(self, data, **kwargs):
        """在外部服务创建 issue"""

    @abstractmethod
    def get_issue(self, issue_id, **kwargs):
        """获取外部 issue 详情"""

    @abstractmethod
    def search_issues(self, query, **kwargs):
        """搜索外部 issue"""

class IssueSyncIntegration(IssueBasicIntegration, ABC):
    outbound_status_key = "sync_status_forward"   # 状态传出
    inbound_status_key = "sync_status_reverse"    # 状态传入
    outbound_assignee_key = "sync_forward_assignment"  # 分配传出
    inbound_assignee_key = "sync_reverse_assignment"   # 分配传入
    comment_key = "sync_comments"                 # 评论同步

    @abstractmethod
    def sync_assignee_outbound(self, external_issue, user, assign=True, **kwargs):
        """将 Sentry 的分配变更同步到外部"""

    @abstractmethod
    def sync_status_outbound(self, external_issue, is_resolved, project_id):
        """将 Sentry 的状态变更同步到外部"""

    @abstractmethod
    def get_resolve_sync_action(self, data: Mapping[str, Any]) -> ResolveSyncAction:
        """根据外部 webhook 数据判断应执行的状态动作"""

    def sync_status_inbound(self, issue_key, data):
        """从外部状态变更触发 Sentry 状态更新（通过异步任务）"""

    def create_comment(self, issue_id, user_id, group_note):
        """在外部 issue 上创建评论"""

    def update_comment(self, issue_id, user_id, group_note):
        """更新外部 issue 上的评论"""

class ResolveSyncAction(enum.Enum):
    NOOP = 0       # 不执行任何操作
    RESOLVE = 1    # 解决 Sentry Issue
    UNRESOLVE = 2  # 取消解决 Sentry Issue

    @classmethod
    def from_resolve_unresolve(cls, should_resolve, should_unresolve):
        if should_resolve and should_unresolve:
            return ResolveSyncAction.NOOP  # 冲突情况不做操作
        if should_resolve:
            return ResolveSyncAction.RESOLVE
        if should_unresolve:
            return ResolveSyncAction.UNRESOLVE
        return ResolveSyncAction.NOOP

class RepositoryIntegration(IntegrationInstallation, ABC):
    @abstractmethod
    def get_repositories(self, query=None, ...) -> list[RepositoryInfo]:
        """获取可访问的仓库列表"""

    @abstractmethod
    def has_repo_access(self, repo: RpcRepository) -> bool:
        """检查对特定仓库的访问权限"""

    @abstractmethod
    def source_url_matches(self, url: str) -> bool:
        """判断 URL 是否属于该集成"""

    @abstractmethod
    def format_source_url(self, repo, filepath, branch) -> str:
        """格式化源码链接（用于 stacktrace linking）"""

    @abstractmethod
    def extract_branch_from_source_url(self, repo, url) -> str:
        """从源码 URL 中提取分支名"""

    @abstractmethod
    def extract_source_path_from_source_url(self, repo, url) -> str:
        """从源码 URL 中提取文件路径"""

    def is_broken_integration_error(self, exc) -> HaltReason | None:
        """判断错误是否表示集成已损坏（被限速/暂停/未授权）"""

class IntegrationNotificationClient:
    def send_notification(self, target: IntegrationNotificationTarget, payload) -> None:
        """向指定目标发送通知"""

    def send_notification_with_threading(
        self, target, payload, threading_context
    ) -> dict[str, Any]:
        """发送带线程上下文的通知"""

class GitHubIntegration(
    RepositoryIntegration[GitHubBaseClient],
    GitHubIssuesSpec,
    GitHubIssueSyncSpec,
    CommitContextIntegration,
    RepoTreesIntegration,
):

def _get_eligible_multi_org_installations(client, owner_orgs):
    installed_orgs = client.get_user_info_installations()
    return [
        {
            "installation_id": str(installation.get("id")),
            "github_account": installation.get("account").get("login"),
            "avatar_url": installation.get("account").get("avatar_url"),
        }
        for installation in installed_orgs["installations"]
        if (installation.get("account").get("login") in owner_orgs
            or installation.get("target_type") == "User")
    ]

def get_repositories(self, query=None, page_number_limit=None,
                     accessible_only=False, use_cache=False,
                     raise_on_page_limit=False, parallel=False):
    # 1. 无查询或 accessible_only: 使用 Installation API 获取全部可访问仓库
    # 2. 有查询 + accessible_only=False: 使用 GitHub Search API
    # 3. use_cache=True: 使用带缓存的批量获取
    # 4. parallel=True: 并发获取多页结果

class SlackIntegration(NotifyBasicMixin, IntegrationInstallation, IntegrationNotificationClient):
    def send_notification(self, target, payload):
        client = self.get_client()
        client.chat_postMessage(
            channel=target.resource_id,
            blocks=payload["blocks"],
            text=payload["text"],
            attachments=payload.get("attachments"),
            unfurl_links=False,
            unfurl_media=False,
        )

class SlackIntegrationProvider(IntegrationProvider):
    identity_oauth_scopes = frozenset([
        "channels:read", "channels:history", "groups:read",
        "users:read", "chat:write", "links:read", "links:write",
        "team:read", "im:read", "im:history", "commands",
        "chat:write.public", "chat:write.customize",
    ])
    staging_oauth_scopes: frozenset[str] = frozenset()  # 暂存新范围
    user_scopes = frozenset(["links:read", "users:read", "users:read.email"])

class SlackRenderable(TypedDict):
    blocks: list[dict[str, Any]]
    text: str
    attachments: NotRequired[list[dict[str, Any]]]

def get_create_issue_config(self, group, user, **kwargs):
    # 1. 获取 Jira project 列表
    # 2. 获取指定 project 的 issue create metadata
    # 3. 构建动态字段：
    #    - select 类型字段 → 查询 allowedValues
    #    - user/team 类型 → 设置 autocomplete URL
    #    - sprint/epic → 提供搜索端点
    #    - textarea 类型 → 使用 textarea 控件
    #    - array 类型 → multiple select
    # 4. 应用排序 (anti_gravity): priority → fixVersions → components → ...

def get_resolve_sync_action(self, data):
    done_statuses = self._get_done_statuses()
    c_from = data["changelog"]["from"]
    c_to = data["changelog"]["to"]
    return ResolveSyncAction.from_resolve_unresolve(
        should_resolve=c_to in done_statuses and c_from not in done_statuses,
        should_unresolve=c_from in done_statuses and c_to not in done_statuses,
    )

CUSTOM_ERROR_MESSAGE_MATCHERS = [
    (re.compile("Team with id '.*' not found.$"), "Team Field"),
    (re.compile(r"Issue does not exist or you do not have permission..."), "Issue"),
]

class VercelIntegrationProvider(IntegrationProvider):
    can_add = False
    can_add_externally = True

class VercelOAuthApiStep(OAuth2ApiStep):
    def extract_code(self, validated_data, pipeline):
        # code 是 marketplace 跳转时注入的初始数据
        return cast(str, pipeline.fetch_state("code"))

def post_install(self, integration, organization, *, extra):
    sentry_app = SentryAppCreator(
        name="Vercel Internal Integration",
        author="Auto-generated by Sentry",
        organization_id=organization.id,
        is_internal=True,
        verify_install=False,
        overview=internal_integration_overview,
        scopes=["org:ci"],
    ).run(user=user)
    SentryAppInstallationForProvider.objects.create(
        sentry_app_installation=sentry_app_installation,
        organization_id=organization.id,
        provider="vercel",
    )

class MsTeamsIntegrationProvider(IntegrationProvider):
    can_add = False
    can_add_externally = True

class MsTeamsInstallParams(TypedDict):
    external_id: str
    external_name: str
    service_url: str
    user_id: str
    conversation_id: str
    tenant_id: str
    installation_type: str

# 验证并解析
class MsTeamsInitialDataSerializer(CamelSnakeSerializer):
    signed_params = CharField(required=True)

    def validate(self, attrs):
        try:
            return unsign(attrs["signed_params"],
                         max_age=INSTALL_EXPIRATION_TIME, salt=SALT)
        except SignatureExpired:
            raise ValidationError("Installation link expired")

def post_install(self, integration, organization, *, extra):
    client = MsTeamsClient(integration)
    card = (build_team_installation_confirmation_message(organization)
            if integration.metadata["installation_type"] == "team"
            else build_personal_installation_confirmation_message())
    client.send_card(extra["conversation_id"], card)

@cell_silo_endpoint
class GitlabWebhookEndpoint(Endpoint):
    _handlers = {
        "Push Hook": PushEventWebhook,
        "Merge Request Hook": MergeEventWebhook,
        "Note Hook": NoteEventWebhook,
        "Issue Hook": IssuesEventWebhook,
    }

    def post(self, request):
        # 1. 从 HTTP_X_GITLAB_TOKEN 提取 external_id 和 secret
        # 2. 查找对应的 Integration
        # 3. 验证 webhook secret
        # 4. 根据 HTTP_X_GITLAB_EVENT 选择 handler
        # 5. 对所有安装了该集成的组织执行 handler
        for install in installs:
            handler = self._handlers[request.META["HTTP_X_GITLAB_EVENT"]]()
            handler(event, integration=integration, organization=organization)
        return HttpResponse(status=204)

def get_gitlab_external_id(request, extra):
    token = request.META["HTTP_X_GITLAB_TOKEN"]
    # e.g. "example.gitlab.com:group-x:webhook_secret"
    instance, group_path, secret = token.split(":")
    external_id = f"{instance}:{group_path}"
    return (external_id, secret)

class BitbucketWebhookEndpoint(Endpoint):
    def post(self, request, organization_id):
        # IP 范围检查
        ip = ipaddress.ip_address(request.META["REMOTE_ADDR"])
        valid_ip = any(ip in ip_range for ip_range in BITBUCKET_IP_RANGES)

        # 签名验证
        secret = integration.metadata["webhook_secret"]
        if not is_valid_signature(request.body, secret, signature):
            raise WebhookInvalidSignatureException()

# URL 配置 (jira/urls.py)
urlpatterns = [
    path("issue-updated/", JiraIssueUpdatedWebhook.as_view(), name="sentry-extensions-jira-issue-updated"),
    path("installed/", JiraInstalledWebhook.as_view(), name="sentry-extensions-jira-installed"),
    path("uninstalled/", JiraUninstalledWebhook.as_view(), name="sentry-extensions-jira-uninstalled"),
]

class GitlabWebhook(SCMWebhook, ABC):
    WEBHOOK_EVENT_PROCESSORS: tuple[WebhookProcessor, ...] = ()

    def _handle(self, integration, event, organization, repo, **kwargs):
        for processor in self.WEBHOOK_EVENT_PROCESSORS:
            try:
                processor(event=event, integration=integration,
                         organization=organization, repo=repo, **kwargs)
            except Exception as e:
                sentry_sdk.capture_exception(e)
                continue  # 一个处理器失败不影响其他

class MergeEventWebhook(GitlabWebhook):
    WEBHOOK_EVENT_PROCESSORS = (
        track_gitlab_contributor_seat_processor,  # 追踪贡献者席位
        track_gitlab_contributor_action_processor,  # 追踪贡献者操作
        handle_merge_request_event,                 # 处理 MR 事件
    )

# IntegrationWebhookEvent 用于记录事件指标
with IntegrationWebhookEvent(
    interaction_type=event_handler.event_type,
    domain=IntegrationDomain.SOURCE_CODE_MANAGEMENT,
    provider_key=event_handler.provider,
).capture():
    event_handler(event, repo=repo, organization=organization)

# webhook_viewer_context 用于关联日志
with webhook_viewer_context(organization.id):
    ...

def _prepare_stacktrace_link(self):
    schema = self.component.app_schema
    uri = schema.get("uri")

    # 将相对 URI 拼接到 webhook_url
    urlparts = list(urlparse(self.install.sentry_app.webhook_url))
    urlparts[2] = str(uri)

    # 注入 installationId 和 projectSlug
    query = {"installationId": self.install.uuid}
    if self.project_slug:
        query["projectSlug"] = self.project_slug
    urlparts[4] = urlencode(query)

    schema.update({"url": urlunparse(urlparts)})

EVENT_EXPANSION = {
    "issue": ["issue.created", "issue.resolved", "issue.ignored", "issue.assigned"],
    "comment": ["comment.created", "comment.updated", "comment.deleted"],
    "error": ["error.created"],
}

class HookService(RpcService):
    def create_service_hook(self, *, application_id, actor_id, installation_id,
                            organization_id, project_ids, events, url):
        """为 Sentry App 创建 ServiceHook"""

    def update_webhook_and_events(self, *, organization_id, application_id,
                                   webhook_url, events):
        """更新所有安装的 Webhook 配置"""

class SentryAppCreator:
    def __init__(self, *, name, author, organization_id, is_internal=False,
                 scopes, webhook_url=None, redirect_url=None,
                 verify_install=True, overview=None, schema=None):
        ...

    def run(self, user) -> SentryApp:
        # 1. 创建 ApiApplication (OAuth 客户端)
        # 2. 创建 SentryApp
        # 3. 创建 SentryAppComponent (如果提供了 schema)
        # 4. 创建 ApiToken (如果是 Internal Integration)
        # 5. 创建 ServiceHook (如果提供了 webhook_url)
        # 6. 自动安装 (如果是 Internal Integration)

class BindingManager:
    BINDINGS = {
        "integration-repository.provider": IntegrationRepositoryProviderManager,
    }

    def __init__(self):
        self._bindings = {k: v() for k, v in self.BINDINGS.items()}

    def add(self, name, binding, **kwargs):
        self._bindings[name].add(binding, **kwargs)

    def get(self, name):
        return self._bindings[name]

# 全局绑定管理器
bindings = BindingManager()

# GitHub 集成在 setup() 中注册 Repository Provider
def setup(self):
    from sentry.plugins.base import bindings
    bindings.add("integration-repository.provider", GitHubRepositoryProvider, id="integrations:github")

class IntegrationRepositoryProvider(RepositoryProvider):
    def get_installation(self, integration_id, organization_id):
        """通过 integration_id 获取 IntegrationInstallation 实例"""

    def get_repository_data(self, organization, config):
        """构建仓库配置数据"""

    def build_repository_config(self, organization, data) -> RepositoryConfig:
        """构建 RepositoryConfig"""
        return {
            "name": data["name"],
            "external_id": data["external_id"],
            "url": data.get("url"),
            "config": data.get("config", {}),
            "integration_id": int(data["integration_id"]),
        }

    def compare_commits(self, repo, start_sha, end_sha):
        """比较两个 commit 之间的提交记录"""

# integration.py

from sentry.integrations.base import (
    FeatureDescription,
    IntegrationFeatures,
    IntegrationMetadata,
)
from django.utils.translation import gettext_lazy as _

DESCRIPTION = """
Connect your Sentry organization to your Tracely workspace. Create and link
Tracely tickets directly from Sentry issues, and synchronize ticket statuses
bidirectionally.
"""

FEATURES = [
    FeatureDescription(
        """
        Create and link Sentry issue groups directly to a Tracely ticket,
        providing a quick way to jump from a Sentry bug to tracked ticket.
        """,
        IntegrationFeatures.ISSUE_BASIC,
    ),
    FeatureDescription(
        """
        Automatically synchronize statuses to and from Tracely. When a ticket
        is marked closed in Tracely, resolve the linked Sentry issue.
        """,
        IntegrationFeatures.ISSUE_SYNC,
    ),
]

metadata = IntegrationMetadata(
    description=DESCRIPTION.strip(),
    features=FEATURES,
    author="Your Company",
    noun=_("Workspace"),
    issue_url="https://github.com/your-org/sentry-tracely/issues/new",
    source_url="https://github.com/your-org/sentry-tracely",
    aspects={},
)

from sentry.integrations.base import IntegrationInstallation
from sentry.integrations.mixins.issues import IssueSyncIntegration, ResolveSyncAction
from sentry.integrations.models.external_issue import ExternalIssue
from sentry.shared_integrations.exceptions import IntegrationError


class TracelyIntegration(IssueSyncIntegration):
    # IssueSyncIntegration 配置键
    comment_key = "sync_comments"
    outbound_status_key = "sync_status_forward"
    inbound_status_key = "sync_status_reverse"
    outbound_assignee_key = "sync_forward_assignment"
    inbound_assignee_key = "sync_reverse_assignment"

    def get_client(self):
        """返回 Tracely API 客户端"""
        return TracelyApiClient(
            base_url=self.model.metadata["base_url"],
            api_key=self.model.metadata["api_key"],
        )

    def get_issue_url(self, key: str) -> str:
        return f"{self.model.metadata['base_url']}/tickets/{key}"

    def get_issue(self, issue_id, **kwargs):
        client = self.get_client()
        try:
            ticket = client.get_ticket(issue_id)
        except ApiError as e:
            self.raise_error(e)
        return {
            "key": ticket["id"],
            "title": ticket["title"],
            "description": ticket["description"],
        }

    def create_issue(self, data, **kwargs):
        client = self.get_client()
        try:
            ticket = client.create_ticket(
                title=data["title"],
                description=data.get("description", ""),
                project=data.get("project"),
            )
        except ApiError as e:
            self.raise_error(e)
        return {
            "key": ticket["id"],
            "title": ticket["title"],
            "description": ticket["description"],
        }

    def search_issues(self, query, **kwargs):
        client = self.get_client()
        results = client.search_tickets(query)
        return [{"label": f"#{t['id']} {t['title']}", "value": t["id"]}
                for t in results]

    def sync_assignee_outbound(self, external_issue, user, assign=True, **kwargs):
        client = self.get_client()
        tracely_user_id = None
        if assign and user:
            tracely_user_id = self._find_tracely_user(user)
        client.assign_ticket(external_issue.key, tracely_user_id)

    def sync_status_outbound(self, external_issue, is_resolved, project_id):
        client = self.get_client()
        new_status = "closed" if is_resolved else "open"
        client.update_ticket_status(external_issue.key, new_status)

    def get_resolve_sync_action(self, data):
        status = data.get("status", "")
        return ResolveSyncAction.from_resolve_unresolve(
            should_resolve=status == "closed",
            should_unresolve=status == "reopened",
        )

    def get_organization_config(self):
        return [
            {
                "name": self.outbound_status_key,
                "type": "choice_mapper",
                "label": "Sync Sentry Status to Tracely",
                "help": "When a Sentry issue changes status...",
                "addButtonText": "Add Tracely Project",
                "addDropdown": {
                    "emptyMessage": "All projects configured",
                    "items": [],  # 动态填充
                },
                "mappedSelectors": {
                    "on_resolve": {"choices": [("closed", "Closed")]},
                    "on_unresolve": {"choices": [("open", "Open")]},
                },
            },
            {"name": self.comment_key, "type": "boolean",
             "label": "Sync Sentry Comments to Tracely"},
        ]

    def create_comment(self, issue_id, user_id, group_note):
        client = self.get_client()
        return client.create_comment(
            issue_id, group_note.data["text"],
        )

    def get_create_issue_config(self, group, user, **kwargs):
        fields = super().get_create_issue_config(group, user, **kwargs)
        # 添加项目选择字段
        client = self.get_client()
        projects = client.get_projects()
        fields.insert(0, {
            "name": "project",
            "label": "Tracely Project",
            "type": "select",
            "choices": [(p["id"], p["name"]) for p in projects],
            "required": True,
        })
        return fields

# client.py

from sentry.shared_integrations.client.base import BaseApiClient
from sentry.shared_integrations.exceptions import ApiError


class TracelyApiClient(BaseApiClient):
    base_url: str
    api_key: str

    def __init__(self, base_url: str, api_key: str):
        super().__init__()
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key

    def _get_headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

    def get_ticket(self, ticket_id: str) -> dict:
        return self.get(f"/api/v1/tickets/{ticket_id}")

    def create_ticket(self, title: str, description: str, project: str) -> dict:
        return self.post("/api/v1/tickets", data={
            "title": title,
            "description": description,
            "project": project,
        })

    def search_tickets(self, query: str) -> list[dict]:
        return self.get("/api/v1/tickets/search", params={"q": query})

    def assign_ticket(self, ticket_id: str, user_id: str | None):
        return self.put(f"/api/v1/tickets/{ticket_id}/assignee", data={
            "user_id": user_id,
        })

    def update_ticket_status(self, ticket_id: str, status: str):
        return self.put(f"/api/v1/tickets/{ticket_id}/status", data={
            "status": status,
        })

    def create_comment(self, ticket_id: str, body: str):
        return self.post(f"/api/v1/tickets/{ticket_id}/comments", data={
            "body": body,
        })

    def get_projects(self) -> list[dict]:
        return self.get("/api/v1/projects")

    def get_user_by_email(self, email: str) -> dict | None:
        users = self.get("/api/v1/users/search", params={"email": email})
        return users[0] if users else None

    def request(self, method, path, **kwargs):
        # 统一错误处理
        try:
            return self._request(method, f"{self.base_url}{path}", **kwargs)
        except Exception as e:
            raise ApiError(str(e))

from sentry.pipeline.types import PipelineStepResult
from sentry.pipeline.views.base import ApiPipelineSteps
from sentry.api.serializers.rest_framework.base import CamelSnakeSerializer
from rest_framework.fields import CharField


class TracelyConfigSerializer(CamelSnakeSerializer):
    base_url = CharField(required=True)
    api_key = CharField(required=True)


class TracelyConfigApiStep:
    step_name = "config"

    def get_step_data(self, pipeline, request):
        return {
            "fields": [
                {"name": "base_url", "label": "Tracely URL",
                 "type": "text", "required": True,
                 "placeholder": "https://your-company.tracely.io"},
                {"name": "api_key", "label": "API Key",
                 "type": "password", "required": True},
            ]
        }

    def get_serializer_cls(self):
        return TracelyConfigSerializer

    def handle_post(self, validated_data, pipeline, request):
        # 验证 API 连通性
        client = TracelyApiClient(
            base_url=validated_data["base_url"],
            api_key=validated_data["api_key"],
        )
        try:
            client.get_projects()
        except ApiError:
            return PipelineStepResult.error(
                "Unable to connect to Tracely. Please verify your URL and API key."
            )

        pipeline.bind_state("base_url", validated_data["base_url"])
        pipeline.bind_state("api_key", validated_data["api_key"])
        return PipelineStepResult.advance()

from sentry.integrations.base import IntegrationProvider, IntegrationData
from sentry.integrations.pipeline import IntegrationPipeline


class TracelyIntegrationProvider(IntegrationProvider):
    key = "tracely"
    name = "Tracely"
    metadata = metadata
    integration_cls = TracelyIntegration
    features = frozenset([
        IntegrationFeatures.ISSUE_BASIC,
        IntegrationFeatures.ISSUE_SYNC,
    ])

    def get_pipeline_api_steps(self) -> ApiPipelineSteps[IntegrationPipeline]:
        return [TracelyConfigApiStep()]

    def build_integration(self, state) -> IntegrationData:
        return {
            "external_id": state["base_url"],
            "name": state["base_url"].split("//")[1].split(".")[0],
            "metadata": {
                "base_url": state["base_url"],
                "api_key": state["api_key"],
            },
        }

    def setup(self):
        """注册 URL 路由"""
        from django.urls import re_path

        from .webhooks import TracelyWebhookEndpoint

        # 在实际项目中，路由注册通过 URL 配置文件完成

from sentry.testutils.cases import TestCase

class TracelyConfigApiStepTest(TestCase):
    def setUp(self):
        self.organization = self.create_organization()
        self.pipeline = self.create_integration_pipeline(
            organization=self.organization,
            provider_key="tracely",
        )

    def test_get_step_data(self):
        step = TracelyConfigApiStep()
        data = step.get_step_data(self.pipeline, self.make_request())
        assert len(data["fields"]) == 2
        assert data["fields"][0]["name"] == "base_url"

    def test_handle_post_success(self):
        step = TracelyConfigApiStep()
        result = step.handle_post(
            {"base_url": "https://test.tracely.io", "api_key": "tk_test123"},
            self.pipeline,
            self.make_request(),
        )
        assert result.action == "advance"
        assert self.pipeline.fetch_state("base_url") == "https://test.tracely.io"

    def test_handle_post_api_error(self):
        step = TracelyConfigApiStep()
        # 使用 mock 模拟 API 失败
        result = step.handle_post(
            {"base_url": "https://invalid.tracely.io", "api_key": "bad_key"},
            self.pipeline,
            self.make_request(),
        )
        assert result.action == "error"

class TracelyIntegrationTest(TestCase):
    def setUp(self):
        self.organization = self.create_organization()
        self.integration = self.create_integration(
            organization=self.organization,
            provider="tracely",
            external_id="test.tracely.io",
            metadata={"base_url": "https://test.tracely.io", "api_key": "tk_test"},
        )
        self.installation = self.integration.get_installation(
            self.organization.id,
        )

    @mock.patch("sentry.integrations.tracely.client.TracelyApiClient.get_ticket")
    def test_get_issue(self, mock_get_ticket):
        mock_get_ticket.return_value = {
            "id": "TCK-123",
            "title": "Test Ticket",
            "description": "A test ticket",
        }
        issue = self.installation.get_issue("TCK-123")
        assert issue["key"] == "TCK-123"
        assert issue["title"] == "Test Ticket"

class TracelyWebhookEndpointTest(TestCase):
    def setUp(self):
        self.organization = self.create_organization()
        self.integration = self.create_integration(
            organization=self.organization,
            provider="tracely",
            external_id="test.tracely.io",
            metadata={"base_url": "https://test.tracely.io",
                       "webhook_secret": "whsec_test123"},
        )

    def test_valid_webhook(self):
        payload = {
            "event": "ticket.closed",
            "ticket": {"id": "TCK-123", "status": "closed"},
        }
        signature = self._compute_signature(payload, "whsec_test123")
        response = self.client.post(
            reverse("sentry-extensions-tracely-webhook"),
            data=json.dumps(payload),
            content_type="application/json",
            HTTP_X_TRACELY_SIGNATURE=signature,
        )
        assert response.status_code == 200

    def test_invalid_signature(self):
        payload = {"event": "ticket.closed"}
        response = self.client.post(
            reverse("sentry-extensions-tracely-webhook"),
            data=json.dumps(payload),
            content_type="application/json",
            HTTP_X_TRACELY_SIGNATURE="invalid",
        )
        assert response.status_code == 401

class MyIntegrationProvider(IntegrationProvider):
    requires_feature_flag = True  # 只有开启了 Flag 的组织才能看到
    # 默认 Flag 名: organizations:integrations-{key}

    # 自定义 Flag 名:
    feature_flag_name = "organizations:my-custom-flag"

def feature_flag_name(f: str | None) -> str | None:
    if f is not None:
        return f"integrations-{f}"
    return None

logger = logging.getLogger("sentry.integrations.tracely")

# 安装尝试
metrics.incr("sentry.integrations.installation_attempt",
             tags={"integration_name": "tracely"})

# 安装完成
metrics.incr("sentry.integrations.installation_finished",
             tags={"integration_name": "tracely"})

# API 错误
metrics.incr("sentry.integrations.tracely.api_error",
             tags={"endpoint": "create_ticket", "status_code": "500"})