main
use queue::Labels;
use sea_orm::DbErr;
use serde_json::Value;
use crate::label;
macro_rules! event_kinds {
($($variant:ident: $kind:literal @ $topic:literal => $tag:literal;)*) => {
/// 业务事件的唯一真源：wire kind、topic area、tag、durable 与串行化粒度都从这里派生。
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum EventKind {
$($variant,)*
}
impl EventKind {
pub const ALL: &'static [EventKind] = &[$(EventKind::$variant,)*];
pub const fn as_str(self) -> &'static str {
match self {
$(Self::$variant => $kind,)*
}
}
pub fn parse(kind: &str) -> Option<Self> {
match kind {
$($kind => Some(Self::$variant),)*
_ => None,
}
}
/// RocketMQ topic 的 area 段：实际 topic 是 `<prefix>-<area>`。
pub const fn topic(self) -> &'static str {
match self {
$(Self::$variant => $topic,)*
}
}
fn only_tag(self) -> &'static [&'static str] {
match self {
$(Self::$variant => &[$tag],)*
}
}
}
};
}
event_kinds! {
AuthRegisterCreate: "auth.register.create" @ "auth" => "register";
AuthLoginCreate: "auth.login.create" @ "auth" => "login";
AuthLogoutDelete: "auth.logout.delete" @ "auth" => "logout";
AuthVerificationTokenCreate: "auth.verification_token.create" @ "auth" => "send-code";
AuthPasswordCreate: "auth.password.create" @ "auth" => "request-reset";
AuthPasswordUpdate: "auth.password.update" @ "auth" => "reset-password";
AuthSshKeyCreate: "auth.ssh_key.create" @ "auth" => "add-sshkey";
AuthSshKeyDelete: "auth.ssh_key.delete" @ "auth" => "delete-sshkey";
AuthAccessKeyCreate: "auth.access_key.create" @ "auth" => "create-accesskey";
AuthAccessKeyDelete: "auth.access_key.delete" @ "auth" => "delete-accesskey";
AuthAccessKeyUsed: "auth.access_key.used" @ "auth" => "use-accesskey";
AuthAppTokenCreate: "auth.app_token.create" @ "auth" => "issue-token";
AuthTokenCreate: "auth.token.create" @ "auth" => "issue-token";
AuthChallengeCreate: "auth.challenge.create" @ "auth" => "challenge";
AuthChallengeUsed: "auth.challenge.used" @ "auth" => "verify-challenge";
EmailDeliveryCreate: "email.delivery.create" @ "auth" => "delivery";
EmailFailureCreate: "email.failure.create" @ "auth" => "failure";
OrganizationLifeCycleCreate: "organization.life_cycle.create" @ "org" => "create";
OrganizationLifeCycleUpdate: "organization.life_cycle.update" @ "org" => "update";
OrganizationLifeCycleDelete: "organization.life_cycle.delete" @ "org" => "delete";
OrganizationInviteCreate: "organization.invite.create" @ "org" => "invite";
OrganizationMemberUpdate: "organization.member.update" @ "org" => "update-member";
OrganizationRoleUpdate: "organization.role.update" @ "org" => "grant";
ChatMessageCreate: "chat.message.create" @ "chat" => "message-create";
ChatMessageUpdate: "chat.message.update" @ "chat" => "message-update";
ChatMessageDelete: "chat.message.delete" @ "chat" => "message-delete";
ChatMessagesDelete: "chat.messages.delete" @ "chat" => "messages-delete";
ChatPinCreate: "chat.pin.create" @ "chat" => "pin-create";
ChatPinDelete: "chat.pin.delete" @ "chat" => "pin-delete";
ChatReactionUpdate: "chat.reaction.update" @ "chat" => "reaction-update";
ChatTypingCreate: "chat.typing.create" @ "chat" => "typing-create";
ChatRoomUpdate: "chat.room.update" @ "chat" => "room-update";
ChatMemberCreate: "chat.member.create" @ "chat" => "member-create";
ChatMemberDelete: "chat.member.delete" @ "chat" => "member-delete";
ChatThreadUpdate: "chat.thread.update" @ "chat" => "thread-update";
ChatVoicePresence: "chat.voice.presence" @ "chat" => "voice-presence";
ChatVoiceHand: "chat.voice.hand" @ "chat" => "voice-hand";
ChatVoiceLayout: "chat.voice.layout" @ "chat" => "voice-layout";
ChatVoiceModerate: "chat.voice.moderate" @ "chat" => "voice-moderate";
ChatMentionCreate: "chat.mention.create" @ "chat" => "mention-create";
ChatReminderCreate: "chat.reminder.create" @ "chat" => "reminder-create";
ChatMembershipCreate: "chat.membership.create" @ "chat" => "membership-create";
ChatReadStateUpdate: "chat.read_state.update" @ "chat" => "read_state-update";
ChatPresenceUpdate: "chat.presence.update" @ "chat" => "presence-update";
NotifyNotificationCreate: "notify.notification.create" @ "notify" => "notification-create";
IssueIssueCreate: "issue.issue.create" @ "issue" => "issue-create";
IssueTitleUpdate: "issue.title.update" @ "issue" => "title-update";
IssueDescriptionUpdate: "issue.description.update" @ "issue" => "description-update";
IssueContentUpdate: "issue.content.update" @ "issue" => "content-update";
IssueMilestoneUpdate: "issue.milestone.update" @ "issue" => "milestone-update";
IssuePriorityUpdate: "issue.priority.update" @ "issue" => "priority-update";
IssueDueUpdate: "issue.due.update" @ "issue" => "due-update";
IssueIssueClose: "issue.issue.close" @ "issue" => "issue-close";
IssueIssueReopen: "issue.issue.reopen" @ "issue" => "issue-reopen";
IssueCommentCreate: "issue.comment.create" @ "issue" => "comment-create";
IssueCommentUpdate: "issue.comment.update" @ "issue" => "comment-update";
IssueCommentDelete: "issue.comment.delete" @ "issue" => "comment-delete";
IssueMemberCreate: "issue.member.create" @ "issue" => "member-create";
IssueMemberDelete: "issue.member.delete" @ "issue" => "member-delete";
IssueLabelCreate: "issue.label.create" @ "issue" => "label-create";
IssueLabelDelete: "issue.label.delete" @ "issue" => "label-delete";
IssueDependencyCreate: "issue.dependency.create" @ "issue" => "dependency-create";
IssueDependencyDelete: "issue.dependency.delete" @ "issue" => "dependency-delete";
IssuePinCreate: "issue.pin.create" @ "issue" => "pin-create";
IssuePinDelete: "issue.pin.delete" @ "issue" => "pin-delete";
IssueRepoRelationCreate: "issue.repo-relation.create" @ "issue" => "repo-relation-create";
IssueRepoRelationDelete: "issue.repo-relation.delete" @ "issue" => "repo-relation-delete";
IssuePullLinkCreate: "issue.pull-link.create" @ "issue" => "pull-link-create";
IssuePullLinkDelete: "issue.pull-link.delete" @ "issue" => "pull-link-delete";
PullPullCreate: "pull.pull.create" @ "pull" => "pull-create";
PullPullUpdate: "pull.pull.update" @ "pull" => "pull-update";
PullPullClose: "pull.pull.close" @ "pull" => "pull-close";
PullPullReopen: "pull.pull.reopen" @ "pull" => "pull-reopen";
PullPullMerge: "pull.pull.merge" @ "pull" => "pull-merge";
PullCommentCreate: "pull.comment.create" @ "pull" => "comment-create";
PullReviewCreate: "pull.review.create" @ "pull" => "review-create";
PullReviewDelete: "pull.review.delete" @ "pull" => "review-delete";
PullReviewRequestCreate: "pull.review-request.create" @ "pull" => "review-request-create";
PullReviewRequestDelete: "pull.review-request.delete" @ "pull" => "review-request-delete";
PullHeadUpdate: "pull.head.update" @ "pull" => "head-update";
PullLabelCreate: "pull.label.create" @ "pull" => "label-create";
PullLabelDelete: "pull.label.delete" @ "pull" => "label-delete";
PullAssigneeCreate: "pull.assignee.create" @ "pull" => "assignee-create";
PullAssigneeDelete: "pull.assignee.delete" @ "pull" => "assignee-delete";
PullLockCreate: "pull.lock.create" @ "pull" => "lock-create";
PullLockDelete: "pull.lock.delete" @ "pull" => "lock-delete";
PullAutoMergeCreate: "pull.auto-merge.create" @ "pull" => "auto-merge-create";
PullAutoMergeDelete: "pull.auto-merge.delete" @ "pull" => "auto-merge-delete";
}
const ORGANIZATION_MEMBER_TAGS: &[&str] = &[
"update-member",
"join",
"leave",
"add-member",
"remove-member",
"set-role",
"ban",
"unban",
"update-ban",
"transfer-owner",
];
const ORGANIZATION_ROLE_TAGS: &[&str] = &[
"grant",
"create-role",
"set-permissions",
"update-role",
"delete-role",
"assign-role",
"unassign-role",
];
impl EventKind {
/// wire kind 的域名段，也是 envelope 的 `source`。
pub fn domain(self) -> &'static str {
self.as_str().split('.').next().unwrap_or_default()
}
/// 该 kind 可能出现的全部 tag，首个是默认值；订阅表达式由此生成。
pub fn tags(self) -> &'static [&'static str] {
match self {
Self::AuthRegisterCreate => &["register", "send-code"],
Self::OrganizationLifeCycleUpdate => &["update", "archive", "restore"],
Self::OrganizationInviteCreate => &["invite", "revoke-invite", "redeem-invite"],
Self::OrganizationMemberUpdate => ORGANIZATION_MEMBER_TAGS,
Self::OrganizationRoleUpdate => ORGANIZATION_ROLE_TAGS,
_ => self.only_tag(),
}
}
/// 实际投递用的 tag：多数 kind 与其他 tag 无关，少数按 payload 细分。
pub fn tag(self, payload: Option<&Value>) -> &'static str {
match self {
Self::AuthRegisterCreate if has_field(payload, "expires_in_secs") => "send-code",
Self::OrganizationLifeCycleUpdate => match action(payload) {
Some("archive") => "archive",
Some("restore") => "restore",
_ => "update",
},
Self::OrganizationInviteCreate => match action(payload) {
Some("revoke") => "revoke-invite",
Some("redeem") => "redeem-invite",
_ => "invite",
},
Self::OrganizationMemberUpdate => match action(payload) {
Some("join_role") => "join",
Some("leave_role") => "leave",
Some("add") => "add-member",
Some("remove") => "remove-member",
Some("set_role") => "set-role",
Some("ban") => "ban",
Some("unban") => "unban",
Some("update_ban") => "update-ban",
Some("transfer_ownership") => "transfer-owner",
_ => "update-member",
},
Self::OrganizationRoleUpdate => match action(payload) {
Some("create") => "create-role",
Some("set_permissions") => "set-permissions",
Some("update") => "update-role",
Some("delete") => "delete-role",
Some("assign") => "assign-role",
Some("unassign") => "unassign-role",
_ => "grant",
},
other => other.tags()[0],
}
}
/// 是否允许写入 outbox：只有持久业务域的事件才进 MQ。
pub fn durable(self) -> bool {
matches!(
self.topic(),
"auth" | "org" | "chat" | "notify" | "issue" | "pull"
)
}
/// 身份域事件走身份 publisher（独立的连接与消费组）。
pub fn identity(self) -> bool {
matches!(self.domain(), "auth" | "email" | "organization")
}
/// 只投递给单个用户的个人事件族（不走资源广播）。
pub const fn personal(self) -> bool {
matches!(
self,
Self::NotifyNotificationCreate
| Self::ChatMentionCreate
| Self::ChatReminderCreate
| Self::ChatMembershipCreate
| Self::ChatReadStateUpdate
| Self::ChatPresenceUpdate
)
}
/// 串行化粒度：同一 `ordering_key` 的事件严格按写入顺序投递。
pub fn ordering_key(self, labels: &Labels) -> Result<String, DbErr> {
if let Some(explicit) = labels
.get(label::ORDERING_KEY)
.filter(|key| !key.is_empty())
{
return Ok(explicit.to_string());
}
let domain = self.domain();
let key = match domain {
"chat" => {
if let Some(uid) = labels.get(label::RECIPIENT_USER_ID) {
return Ok(format!("chat:user:{uid}"));
}
if let Some(room) = labels.get(label::ROOM_ID) {
return Ok(format!("chat:room:{room}"));
}
labels.get(label::USER_ID)
}
"notify" => {
if let Some(uid) = labels
.get(label::RECIPIENT_USER_ID)
.or_else(|| labels.get(label::USER_ID))
{
return Ok(format!("notify:user:{uid}"));
}
None
}
"pull" => labels
.get("pull_id")
.or_else(|| labels.get(label::PULL_REQUEST_ID))
.or_else(|| labels.get(label::REPOSITORY_ID)),
"issue" => labels
.get(label::ISSUE_ID)
.or_else(|| labels.get(label::BOARD_ID))
.or_else(|| labels.get(label::PULL_REQUEST_ID))
.or_else(|| labels.get(label::GUILD_ID)),
"organization" => labels.get(label::ORGANIZATION_ID),
"auth" => labels
.get(label::USER_ID)
.or_else(|| labels.get(label::REQUEST_ID)),
"email" => labels.get(label::REQUEST_ID),
_ => None,
};
key.filter(|key| !key.is_empty())
.map(|key| format!("{domain}:{key}"))
.ok_or_else(|| {
DbErr::Custom(format!(
"event `{}` requires a user, organization or request id label",
self.as_str()
))
})
}
}
impl std::fmt::Display for EventKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
fn action(payload: Option<&Value>) -> Option<&str> {
payload
.and_then(|value| value.get("action"))
.and_then(Value::as_str)
}
fn has_field(payload: Option<&Value>, key: &str) -> bool {
payload.is_some_and(|value| value.get(key).is_some_and(|value| !value.is_null()))
}