打印箱唛修改尾程垮桥事件提交

2026-08-13 17:13:58 +08:00
parent b004edaf9e
commit c43d581ba6
59 changed files with 1609 additions and 96 deletions
@@ -0,0 +1,22 @@
+package com.baiyue.iam.securityevent;
+
+import jakarta.annotation.Resource;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import java.util.UUID;
+
+@Component
+@ConditionalOnProperty(name="baiyue.security-event.publisher-enabled",havingValue="true")
+public class SecurityEventOutboxRelay {
+    @Resource private SecurityEventRepository repository;
+    @Resource private RocketMQTemplate rocketMQ;
+    @Value("${baiyue.security-event.topic:baiyue-security-event-v1}") private String topic;
+    private final String owner=UUID.randomUUID().toString();
+    @Scheduled(fixedDelayString="${baiyue.security-event.poll-interval-ms:5000}")
+    public void relay(){for(String id:repository.pendingIds(50)){if(!repository.claim(id,owner))continue;var row=repository.find(id);try{rocketMQ.syncSend(topic+":"+row.eventType(),MessageBuilder.withPayload(row.payload()).setHeader(RocketMQHeaders.KEYS,row.eventId()).setHeader("eventId",row.eventId()).build(),3000);repository.sent(id,owner);}catch(RuntimeException ex){repository.failed(id,owner,row.attempts(),ex.getMessage());}}}
+}
@@ -0,0 +1,25 @@
+package com.baiyue.iam.securityevent;
+
+import jakarta.annotation.Resource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Repository;
+import java.util.List;
+
+@Repository
+public class SecurityEventRepository {
+    @Resource private JdbcTemplate jdbc;
+    public long incrementPermissionVersion(long authUserId) {
+        jdbc.update("INSERT INTO iam_security_subject_version(auth_user_id,permission_version) VALUES (?,1) ON DUPLICATE KEY UPDATE permission_version=permission_version+1", authUserId);
+        return jdbc.queryForObject("SELECT permission_version FROM iam_security_subject_version WHERE auth_user_id=?", Long.class, authUserId);
+    }
+    public void append(String id,String type,String aggregateType,String aggregateId,String payload) {
+        jdbc.update("INSERT INTO iam_security_event_outbox(event_id,event_type,aggregate_type,aggregate_id,payload_json,status,attempt_count,next_retry_at) VALUES (?,?,?,?,CAST(? AS JSON),'PENDING',0,CURRENT_TIMESTAMP(3))", id,type,aggregateType,aggregateId,payload);
+    }
+    public List<String> pendingIds(int limit) { return jdbc.queryForList("SELECT event_id FROM iam_security_event_outbox WHERE (status='PENDING' AND next_retry_at<=CURRENT_TIMESTAMP(3)) OR (status='SENDING' AND lease_until<=CURRENT_TIMESTAMP(3)) ORDER BY created_at LIMIT ?", String.class,limit); }
+    public boolean claim(String id,String owner) { return jdbc.update("UPDATE iam_security_event_outbox SET status='SENDING',lease_owner=?,lease_until=DATE_ADD(CURRENT_TIMESTAMP(3),INTERVAL 60 SECOND),attempt_count=attempt_count+1 WHERE event_id=? AND ((status='PENDING' AND next_retry_at<=CURRENT_TIMESTAMP(3)) OR (status='SENDING' AND lease_until<=CURRENT_TIMESTAMP(3)))",owner,id)==1; }
+    public Row find(String id) { return jdbc.queryForObject("SELECT event_id,event_type,payload_json,attempt_count FROM iam_security_event_outbox WHERE event_id=?",(rs,n)->new Row(rs.getString(1),rs.getString(2),rs.getString(3),rs.getInt(4)),id); }
+    public void sent(String id,String owner) { jdbc.update("UPDATE iam_security_event_outbox SET status='SENT',sent_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL,last_error=NULL WHERE event_id=? AND lease_owner=?",id,owner); }
+    public void failed(String id,String owner,int attempts,String error) { jdbc.update("UPDATE iam_security_event_outbox SET status=IF(attempt_count>=8,'DEAD','PENDING'),next_retry_at=DATE_ADD(CURRENT_TIMESTAMP(3),INTERVAL ? SECOND),lease_owner=NULL,lease_until=NULL,last_error=? WHERE event_id=? AND lease_owner=?",Math.min(900,5*(1<<Math.min(attempts,7))),safe(error),id,owner); }
+    private String safe(String value) { if(value==null)return "投递失败"; return value.length()<=1000?value:value.substring(0,1000); }
+    public record Row(String eventId,String eventType,String payload,int attempts) { }
+}
@@ -0,0 +1,23 @@
+package com.baiyue.iam.securityevent;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+
+@Service
+public class SecurityEventService {
+    public static final String USER_STATUS_CHANGED="IAM_USER_STATUS_CHANGED_V1";
+    public static final String ACCOUNT_STATUS_CHANGED="IAM_ACCOUNT_STATUS_CHANGED_V1";
+    public static final String PERMISSION_VERSION_CHANGED="IAM_PERMISSION_VERSION_CHANGED_V1";
+    @Resource private SecurityEventRepository repository;
+    @Resource private ObjectMapper json;
+    public void userStatus(long authUserId,long tenantId,long membershipId,String status){append(USER_STATUS_CHANGED,"USER",authUserId,Map.of("authUserId",authUserId,"tenantId",tenantId,"membershipId",membershipId,"status",status));}
+    public void accountStatus(long accountId,String status){append(ACCOUNT_STATUS_CHANGED,"ACCOUNT",accountId,Map.of("accountId",accountId,"status",status));}
+    public long permissionsChanged(long authUserId,long tenantId,long membershipId,String reason){long version=repository.incrementPermissionVersion(authUserId);append(PERMISSION_VERSION_CHANGED,"AUTHORIZATION",authUserId,Map.of("authUserId",authUserId,"tenantId",tenantId,"membershipId",membershipId,"permissionVersion",version,"reason",reason));return version;}
+    private void append(String type,String aggregateType,long aggregateId,Map<String,Object> fields){String id=UUID.randomUUID().toString();Map<String,Object> payload=new LinkedHashMap<>();payload.put("eventId",id);payload.put("eventType",type);payload.put("schemaVersion",1);payload.put("occurredAt",Instant.now().toString());payload.putAll(fields);try{repository.append(id,type,aggregateType,Long.toString(aggregateId),json.writeValueAsString(payload));}catch(JsonProcessingException ex){throw new IllegalStateException("安全事件序列化失败",ex);}}
+}
@@ -0,0 +1,21 @@
+USE baiyue_iam;
+
+CREATE TABLE IF NOT EXISTS iam_security_subject_version (
+  auth_user_id BIGINT NOT NULL COMMENT '认证账号ID',
+  permission_version BIGINT NOT NULL DEFAULT 0 COMMENT '权限版本，角色或权限变化时单调递增',
+  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY (auth_user_id)
+) ENGINE=InnoDB COMMENT='安全主体权限版本';
+
+CREATE TABLE IF NOT EXISTS iam_security_event_outbox (
+  event_id VARCHAR(64) NOT NULL COMMENT '全局事件ID，消费者幂等键', event_type VARCHAR(64) NOT NULL,
+  aggregate_type VARCHAR(32) NOT NULL COMMENT 'USER、ACCOUNT、AUTHORIZATION', aggregate_id VARCHAR(64) NOT NULL,
+  payload_json JSON NOT NULL COMMENT '不含密码、令牌和权限全集的安全事件',
+  status VARCHAR(16) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING、SENDING、SENT、DEAD',
+  attempt_count INT NOT NULL DEFAULT 0, next_retry_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+  lease_owner VARCHAR(128) NULL, lease_until DATETIME(3) NULL, last_error VARCHAR(1000) NULL,
+  sent_at DATETIME(3) NULL, created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY (event_id), KEY idx_security_outbox_publish (status, next_retry_at, lease_until, created_at),
+  KEY idx_security_outbox_aggregate (aggregate_type, aggregate_id, created_at)
+) ENGINE=InnoDB COMMENT='IAM可靠安全事件本地发送箱';