lxq/mydreamjob-server/src/main/java/com/mydreamjob/risk/RiskEventDetector.java

package com.mydreamjob.risk;

import com.mydreamjob.audit.AuditContext;
import com.mydreamjob.audit.AuditLog;
import com.mydreamjob.audit.AuditLogRepository;
import com.mydreamjob.audit.AuditService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.Clock;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 风险事件扫描器（V37）。
 *
 * <p>{@code scanOnce()} 为可直接单测的方法；{@code @Scheduled} 只调用它。扫描基于既有
 * {@code audit_logs}，仅消费 {@code REPORT_RATE_LIMITED}/{@code BLOCKED_SEND_PROHIBITED} 两个
 * 白名单 action，且 user_id 非空。整批成功后才推进 checkpoint；任一处理失败整体回滚。</p>
 *
 * <p>去刷屏：仅当窗口内命中数恰等于 thresholdCount 时创建事件；阈值之后同窗口的更多命中不再
 * 创建新事件。事件插入受唯一键 {@code (rule_config_id, source_audit_log_id)} 与 existsBy 防重保护。</p>
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RiskEventDetector {

    public static final int BATCH_SIZE = 500;
    public static final Set<String> ALLOWED_ACTIONS =
            Set.of("REPORT_RATE_LIMITED", "BLOCKED_SEND_PROHIBITED");

    private static final String AUDIT_DETECTED = "RISK_EVENT_DETECTED";

    private final ScoringRuleConfigRepository ruleRepository;
    private final ScoringRiskEventRepository eventRepository;
    private final ScoringRiskScanCheckpointRepository checkpointRepository;
    private final AuditLogRepository auditLogRepository;
    private final AuditService auditService;
    private final Clock clock;

    /**
     * 执行一次风险扫描。返回本轮新建的风险事件数。
     */
    @Transactional
    public int scanOnce() {
        ScoringRiskScanCheckpoint checkpoint = checkpointRepository
                .findByCheckpointKeyForUpdate(ScoringRiskScanCheckpoint.KEY_RISK_EVENT_DETECTOR)
                .orElseGet(this::createCheckpoint);

        long lastId = checkpoint.getLastAuditLogId();
        List<AuditLog> batch = auditLogRepository
                .findByIdAfterOrderByIdAsc(lastId, PageRequest.of(0, BATCH_SIZE));
        if (batch.isEmpty()) {
            return 0;
        }

        Map<String, ScoringRuleConfig> ruleByAction = ruleRepository.findAllByEnabledTrue().stream()
                .collect(Collectors.toMap(ScoringRuleConfig::getSourceAction, Function.identity()));

        LocalDateTime now = LocalDateTime.now(clock);
        long maxId = lastId;
        int created = 0;

        for (AuditLog a : batch) {
            if (a.getId() > maxId) {
                maxId = a.getId();
            }
            // 仅处理带用户主体且 action 在规则白名单内的审计
            if (a.getUserId() == null || a.getCreatedAt() == null) {
                continue;
            }
            ScoringRuleConfig rule = ruleByAction.get(a.getAction());
            if (rule == null) {
                // 规则已禁用或非白名单 action：审计照常越过 checkpoint；重新启用不回放历史
                continue;
            }

            LocalDateTime winStart = a.getCreatedAt().minusMinutes(rule.getWindowMinutes());
            long occurrence = auditLogRepository.countByUserIdAndActionAndCreatedAtBetween(
                    a.getUserId(), a.getAction(), winStart, a.getCreatedAt());

            // 仅恰好等于阈值才生成事件（防止阈值后每次操作刷屏）
            if (occurrence == rule.getThresholdCount().longValue()
                    && !eventRepository.existsByRuleIdAndSourceAuditLogId(rule.getId(), a.getId())) {
                ScoringRiskEvent ev = ScoringRiskEvent.builder()
                        .rule(rule)
                        .subjectUserId(a.getUserId())
                        .sourceAuditLogId(a.getId())
                        .sourceAction(rule.getSourceAction())
                        .severity(rule.getSeverity())
                        .windowStartedAt(winStart)
                        .windowEndedAt(a.getCreatedAt())
                        .occurrenceCount((int) occurrence)
                        .status(ScoringRiskEvent.STATUS_OPEN)
                        .detectedAt(now)
                        .build();
                // 使用 saveAndFlush 的返回对象，确保事件 ID 已生成（IDENTITY 生成策略下必须用返回值，
                // 否则 ev.getId() 可能仍为 null，导致检测审计 entity_id/details.riskEventId 无法定位）。
                ScoringRiskEvent savedEvent = eventRepository.saveAndFlush(ev);
                created++;

                auditService.log(null, AUDIT_DETECTED, "SCORING_RISK_EVENT", savedEvent.getId(),
                        detectedAuditDetails(savedEvent), AuditContext.builder().build());
            }
        }

        checkpoint.setLastAuditLogId(maxId);
        checkpointRepository.save(checkpoint);
        return created;
    }

    private ScoringRiskScanCheckpoint createCheckpoint() {
        ScoringRiskScanCheckpoint cp = ScoringRiskScanCheckpoint.builder()
                .checkpointKey(ScoringRiskScanCheckpoint.KEY_RISK_EVENT_DETECTOR)
                .lastAuditLogId(0L)
                .build();
        return checkpointRepository.saveAndFlush(cp);
    }

    /**
     * RISK_EVENT_DETECTED 审计 details（JSON 形式规避 AuditService 对 {@code code=} 的脱敏误伤，
     * 只记录非敏感追踪字段，不含用户身份/原始审计 details/IP/UA/凭据）。
     */
    private static String detectedAuditDetails(ScoringRiskEvent ev) {
        return "{\"riskEventId\":" + ev.getId()
                + ",\"ruleCode\":\"" + ev.getRule().getRuleCode() + "\""
                + ",\"severity\":\"" + ev.getSeverity() + "\""
                + ",\"occurrenceCount\":" + ev.getOccurrenceCount() + "}";
    }
}