S8WatchSchedulerService.cs 73 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579
  1. using Admin.NET.Plugin.AiDOP.Const.S8;
  2. using Admin.NET.Plugin.AiDOP.Entity.S8;
  3. using Admin.NET.Plugin.AiDOP.Infrastructure.S8;
  4. using Admin.NET.Plugin.AiDOP.Service.S8.Rules;
  5. using Admin.NET.Plugin.AiDOP.Service.S8.Rules.DataAccess;
  6. using Admin.NET.Plugin.AiDOP.Service.S8.Rules.Definitions;
  7. using Microsoft.Extensions.Logging;
  8. using SqlSugar;
  9. using System.Globalization;
  10. using System.Text.Json;
  11. namespace Admin.NET.Plugin.AiDOP.Service.S8;
  12. /// <summary>
  13. /// 监视规则轮询调度服务(首轮存根)。
  14. /// 后续接入 Admin.NET 定时任务机制后,由调度器周期调用 <see cref="RunOnceAsync"/>,
  15. /// 按各规则的 PollIntervalSeconds 逐条评估并生成异常记录。
  16. /// </summary>
  17. public class S8WatchSchedulerService : ITransient
  18. {
  19. private readonly SqlSugarRepository<AdoS8WatchRule> _ruleRep;
  20. private readonly SqlSugarRepository<AdoS8Exception> _exceptionRep;
  21. private readonly SqlSugarRepository<AdoS8ExceptionType> _exceptionTypeRep;
  22. private readonly S8NotificationService _notificationService;
  23. private readonly S8NotificationLayerResolver _notificationLayerResolver;
  24. private readonly S8ImpactMetricsService _impactMetricsService;
  25. private readonly S8ManualReportService _manualReportService;
  26. // S8-STEP6E:Watch 主链的业务建单边界(唯一最窄 seam)。仅 :1482 的自动建单走它;
  27. // legacy debug 主链的 CreateFromWatchAsync 保持直连 _manualReportService,行为不变。
  28. private readonly IS8ExceptionReportDispatcher _reportDispatcher;
  29. private readonly S8TimeoutRuleEvaluator _timeoutEvaluator;
  30. private readonly S8ShortageRuleEvaluator _shortageEvaluator;
  31. private readonly S8OutOfRangeRuleEvaluator _outOfRangeEvaluator;
  32. private readonly ILogger<S8WatchSchedulerService> _logger;
  33. private readonly SqlSugarRepository<AdoS8DetectionLog> _detectionLogRep;
  34. private readonly SqlSugarRepository<AdoS8RuleDetectionState> _detectionStateRep;
  35. private readonly IS8RuleCatalog _ruleCatalog;
  36. // S8-RULE-READINESS-1:与 Enable / RunNow 共用的就绪门禁。
  37. private readonly S8RuleReadinessGate _readinessGate;
  38. private const string DetectionTriggerSource = "WATCH_SCHEDULER";
  39. private const string DetectResultCreated = "CREATED";
  40. private const string DetectResultRefreshed = "REFRESHED";
  41. private const string DetectResultRecovered = "RECOVERED";
  42. /// <summary>系统恢复时间线的动作码。与人工动作码同一命名空间,但永远不带操作人。</summary>
  43. internal const string SystemRecoveredActionCode = "SYSTEM_RECOVERED";
  44. private const string DetectResultNoHit = "NO_HIT";
  45. private const string DetectResultEvaluateFailed = "EVALUATE_FAILED";
  46. // S8-DETECTION-LOG-WRITE-REDUCE-P1-1:REFRESHED 明细写入限频窗口。
  47. // 同 (tenant, factory, rule_code, dedup_key) 在 10 分钟内最多写一条 REFRESHED detection_log;
  48. // 业务侧 RefreshDetectionAsync / BackfillLegacyExceptionAsync 不受限频影响。
  49. private const int RefreshedRateLimitMinutes = 10;
  50. private const string DefaultTriggerType = "VALUE_DEVIATION";
  51. private const string SqlDataSourceType = "SQL";
  52. // G01-05 未闭环状态集合:复用自 S8ExceptionService 当前 pendingStatuses 事实口径
  53. // (见 S8ExceptionService.GetPagedAsync 中 pendingStatuses 的定义,两处必须保持一致)。
  54. // 这不是“自定义未闭环集合”;若现有口径调整,两处需同步修改。
  55. private static readonly string[] UnclosedExceptionStatuses =
  56. { "NEW", "ASSIGNED", "IN_PROGRESS", "PENDING_VERIFICATION" };
  57. public S8WatchSchedulerService(
  58. SqlSugarRepository<AdoS8WatchRule> ruleRep,
  59. SqlSugarRepository<AdoS8Exception> exceptionRep,
  60. SqlSugarRepository<AdoS8ExceptionType> exceptionTypeRep,
  61. S8NotificationService notificationService,
  62. S8NotificationLayerResolver notificationLayerResolver,
  63. S8ImpactMetricsService impactMetricsService,
  64. S8ManualReportService manualReportService,
  65. IS8ExceptionReportDispatcher reportDispatcher,
  66. S8TimeoutRuleEvaluator timeoutEvaluator,
  67. S8ShortageRuleEvaluator shortageEvaluator,
  68. S8OutOfRangeRuleEvaluator outOfRangeEvaluator,
  69. ILogger<S8WatchSchedulerService> logger,
  70. SqlSugarRepository<AdoS8DetectionLog> detectionLogRep,
  71. SqlSugarRepository<AdoS8RuleDetectionState> detectionStateRep,
  72. IS8RuleCatalog ruleCatalog,
  73. S8RuleReadinessGate readinessGate)
  74. {
  75. _ruleRep = ruleRep;
  76. _exceptionRep = exceptionRep;
  77. _exceptionTypeRep = exceptionTypeRep;
  78. _notificationService = notificationService;
  79. _notificationLayerResolver = notificationLayerResolver;
  80. _impactMetricsService = impactMetricsService;
  81. _manualReportService = manualReportService;
  82. _reportDispatcher = reportDispatcher;
  83. _timeoutEvaluator = timeoutEvaluator;
  84. _shortageEvaluator = shortageEvaluator;
  85. _outOfRangeEvaluator = outOfRangeEvaluator;
  86. _logger = logger;
  87. _detectionLogRep = detectionLogRep;
  88. _detectionStateRep = detectionStateRep;
  89. _ruleCatalog = ruleCatalog;
  90. _readinessGate = readinessGate;
  91. }
  92. /// <summary>
  93. /// S8-TENANT-ONLY-BATCH5:调度遍历改为 <b>Tenant → Rule</b>。
  94. ///
  95. /// <para>原实现返回 DISTINCT (tenant_id, factory_id),于是同一租户在 N 个工厂下会被 tick N 次;
  96. /// 而规则的运行策略按 Batch 2 的口径每租户只有一条,重复 tick 只会互相抢同一把租约。</para>
  97. ///
  98. /// <para>同时删掉 <c>AND r.factory_id &gt; 0</c>:Batch 2 供给出来的运行策略行
  99. /// <c>factory_id</c> 恒为 0,那条谓词会让它们**永远不被调度、且不报任何错**
  100. /// —— 页面显示已启用,实际从不执行。这是本批必须与供给同时闭合的缺口。</para>
  101. /// </summary>
  102. public async Task<List<long>> ListEnabledTenantsAsync()
  103. {
  104. return await _ruleRep.Context.Ado.SqlQueryAsync<long>(
  105. """
  106. SELECT DISTINCT r.tenant_id
  107. FROM ado_s8_watch_rule r
  108. INNER JOIN SysTenant t ON t.Id = r.tenant_id AND t.Status = 1
  109. WHERE r.enabled = 1
  110. AND r.tenant_id > 0
  111. ORDER BY r.tenant_id
  112. """);
  113. }
  114. // 取任意一条匹配的未闭环异常 Id 作为“是否存在重复单”的拦截依据。
  115. // 首版只需要“存在性”,不关心“最早 / 最新”;不在 G01-05 处理排序语义。
  116. /// <summary>
  117. /// 自动建单入口(debug run-once 使用;生产调度走 <see cref="RunDispatchTickAsync"/>)。
  118. ///
  119. /// S8-LEGACY-SQL-RESIDUAL-CLEANUP-3:原先此处还有一段旧 AlertRule 兼容主链
  120. /// (EvaluateDedupAsync → EvaluateHitsAsync → QueryDeviceRowsAsync → 直连 SQL),
  121. /// 该链自行开 SqlSugarScope 执行 rule.expression,
  122. /// 且只装载 rule_type 为空的未分类历史规则 + 需同场景恰好一条可运行 AlertRule。
  123. /// 经证实其唯一入口是 404 门禁的 debug controller,且 LoadExecutionRulesAsync 的过滤条件
  124. /// 在当前数据下恒返回空,已整体删除。本方法现在只保留三类正式 evaluator 路径。
  125. /// </summary>
  126. public async Task<List<S8WatchCreationResult>> CreateExceptionsAsync(long tenantId)
  127. {
  128. var results = new List<S8WatchCreationResult>();
  129. // R6 RunId:本次 CreateExceptionsAsync 调用对应的统一关联 id,落入 detection_log。
  130. var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
  131. results.AddRange(await ProcessRulesByTypeAsync(tenantId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, runId));
  132. results.AddRange(await ProcessRulesByTypeAsync(tenantId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, runId));
  133. results.AddRange(await ProcessRulesByTypeAsync(tenantId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, runId));
  134. return results;
  135. }
  136. /// <summary>
  137. /// R2 TIMEOUT 类规则主链:薄包装,复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
  138. /// </summary>
  139. public Task<List<S8WatchCreationResult>> ProcessTimeoutRulesAsync(long tenantId) =>
  140. ProcessRulesByTypeAsync(tenantId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  141. /// <summary>
  142. /// R3 SHORTAGE 类规则主链:薄包装,复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
  143. /// </summary>
  144. public Task<List<S8WatchCreationResult>> ProcessShortageRulesAsync(long tenantId) =>
  145. ProcessRulesByTypeAsync(tenantId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  146. /// <summary>
  147. /// R3-OUT_OF_RANGE-REWRITE-1:OUT_OF_RANGE 类规则主链。
  148. /// 复用 <see cref="ProcessRulesByTypeAsync"/>,并对历史 dedup_key=NULL 的旧记录做 compat fallback:
  149. /// (source_rule_id=rule.Id AND related_object_code=hit AND status!=CLOSED AND dedup_key IS NULL AND is_deleted=0)
  150. /// 命中则 backfill 6 列,避免重复建单。RunId 由内部生成。
  151. /// </summary>
  152. public Task<List<S8WatchCreationResult>> ProcessOutOfRangeRulesAsync(long tenantId) =>
  153. ProcessRulesByTypeAsync(tenantId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  154. /// <summary>
  155. /// R2/R3 通用规则主链。S8-SCHED-CLEANUP-LEGACY-PATH-1:本方法已收敛为 thin wrapper,
  156. /// 单规则处理(evaluator → reconcile → hit 循环 → CREATED/REFRESHED/NO_HIT/EVALUATE_FAILED 日志)
  157. /// 全部下沉至 <see cref="ProcessSingleRuleAsync"/>,与 Job tick 路径
  158. /// (<see cref="RunDispatchTickAsync"/> → <see cref="RunSingleRuleAsync"/> → ProcessSingleRuleAsync)
  159. /// 共享同一份逻辑,避免双维护。
  160. ///
  161. /// 调用方仅有 <see cref="CreateExceptionsAsync"/>(debug run-once / Process*RulesAsync 公共薄包装)。
  162. /// debug run-once 不持 lease(保留"手动立即跑"语义),但本方法在每条规则结束时通过
  163. /// <see cref="ApplyRunOnceCompletionAsync"/> 同步更新 watch_rule.last_run_at / last_status /
  164. /// last_error / last_duration_ms / last_run_id / consecutive_failure_count,缩小 run-once 与
  165. /// Job tick 之间的 last_* 状态分裂;不动 lock_token / running_started_at / paused_until,避免
  166. /// 与正在持锁运行的 Job 撕扯。
  167. /// </summary>
  168. private async Task<List<S8WatchCreationResult>> ProcessRulesByTypeAsync(
  169. long tenantId, IS8RuleEvaluator evaluator, string ruleType, string runId)
  170. {
  171. var aggregate = new List<S8WatchCreationResult>();
  172. var rules = await _ruleRep.AsQueryable()
  173. .Where(x => x.TenantId == tenantId
  174. && x.Enabled
  175. && x.RuleType == ruleType)
  176. .ToListAsync();
  177. if (rules.Count == 0) return aggregate;
  178. foreach (var rule in rules.OrderBy(x => x.Id))
  179. {
  180. // S8-RULE-GOVERNANCE-BATCH1:无代码定义的规则不进入执行链(与 Job tick 路径同一口径)。
  181. if (!_ruleCatalog.IsDefined(rule.RuleCode))
  182. {
  183. _logger.LogWarning(
  184. "run_once_skip_no_definition ruleId={RuleId} ruleCode={RuleCode}", rule.Id, rule.RuleCode);
  185. aggregate.Add(BuildSkipResult(rule, S8RuleCatalog.ReasonNotFound, null));
  186. continue;
  187. }
  188. // S8-RUN-ONCE-LEASE-AWARENESS-1:debug run-once 不持 lease,但若该 rule 已被 Scheduler Job
  189. // 通过 PickReadyRulesAsync 抢锁(lock_token 非空且 lock_until > now),run-once 跳过该 rule,
  190. // 不写 last_*、不写 detection_log、不动锁,以避免与 Job tick 并发评估同一 rule 而互相覆盖运行态。
  191. // 锁状态用 ToListAsync 的初始快照判定;run-once 与 Job 的微秒级竞速无法在不抢 lease 的前提下
  192. // 完全消除(已登记为 follow-up 风险)。
  193. if (!string.IsNullOrEmpty(rule.LockToken)
  194. && rule.LockUntil.HasValue
  195. && rule.LockUntil.Value > DateTime.Now)
  196. {
  197. _logger.LogInformation(
  198. "run_once_skip_locked ruleId={RuleId} ruleCode={RuleCode} ruleType={RuleType} lockToken={Token} lockedBy={By} lockUntil={Until}",
  199. rule.Id, rule.RuleCode, ruleType, rule.LockToken, rule.LockedBy, rule.LockUntil);
  200. aggregate.Add(BuildSkipResult(rule, "rule_locked_by_scheduler", null));
  201. continue;
  202. }
  203. var sw = System.Diagnostics.Stopwatch.StartNew();
  204. S8RuleRunResult completion;
  205. try
  206. {
  207. var ruleResults = await ProcessSingleRuleAsync(tenantId, rule, ruleType, evaluator, runId);
  208. aggregate.AddRange(ruleResults);
  209. completion = new S8RuleRunResult
  210. {
  211. Success = true,
  212. Stats = new S8RuleRunStats
  213. {
  214. Hits = ruleResults.Count,
  215. Created = ruleResults.Count(r => r.Created),
  216. Refreshed = ruleResults.Count(r => r.Reason == "duplicate_pending"),
  217. Pending = ruleResults.Count(r => r.Reason == "antiflap_pending_hit"),
  218. Failed = ruleResults.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
  219. }
  220. };
  221. }
  222. catch (Exception ex)
  223. {
  224. // ProcessSingleRuleAsync 在 evaluator 抛 S8RuleEvaluatorException 时已写 EVALUATE_FAILED
  225. // detection_log 后再 throw(供 RunSingleRuleAsync 标 Success=false)。本路径无 lease,
  226. // 吞异常以保留"逐规则失败不影响其他规则"的旧 ProcessRulesByTypeAsync 语义。
  227. _logger.LogWarning(ex, "process_rule_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  228. aggregate.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
  229. completion = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
  230. }
  231. sw.Stop();
  232. try
  233. {
  234. await ApplyRunOnceCompletionAsync(rule, completion, (int)sw.ElapsedMilliseconds, runId);
  235. }
  236. catch (Exception ex)
  237. {
  238. _logger.LogWarning(ex, "run_once_completion_write_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  239. }
  240. }
  241. return aggregate;
  242. }
  243. /// <summary>
  244. /// S8-SCHED-CLEANUP-LEGACY-PATH-1:debug run-once 的 last_* 状态回写(无 lease 版)。
  245. /// 与 <see cref="OnRuleCompletedAsync"/> 的语义平行,差异在于:
  246. /// - 不要求 lock_token 匹配(run-once 不持 lease)
  247. /// - 不写 lock_token / locked_by / lock_until / running_started_at(不与 Job lease 撕扯)
  248. /// - 不做 consecutive_failure_count 阈值的 paused_until 自动暂停(debug 路径不应自动暂停 demo rule)
  249. /// 写入:last_run_at / next_run_at / last_status / last_error / last_duration_ms / last_run_id /
  250. /// consecutive_failure_count / updated_at。
  251. /// </summary>
  252. private async Task ApplyRunOnceCompletionAsync(AdoS8WatchRule rule, S8RuleRunResult result, int durationMs, string runId)
  253. {
  254. var now = DateTime.Now;
  255. var nextRunAt = now.AddSeconds(NormalizePollInterval(rule.PollIntervalSeconds));
  256. if (result.Success)
  257. {
  258. await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  259. .SetColumns(x => new AdoS8WatchRule
  260. {
  261. LastRunAt = now,
  262. NextRunAt = nextRunAt,
  263. LastStatus = "SUCCESS",
  264. LastError = null,
  265. LastDurationMs = durationMs,
  266. LastRunId = runId,
  267. ConsecutiveFailureCount = 0,
  268. UpdatedAt = now
  269. })
  270. .Where(x => x.Id == rule.Id)
  271. .ExecuteCommandAsync();
  272. }
  273. else
  274. {
  275. var errorTrunc = Truncate(result.ErrorMessage, 500);
  276. await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  277. .SetColumns(x => new AdoS8WatchRule
  278. {
  279. LastRunAt = now,
  280. NextRunAt = nextRunAt,
  281. LastStatus = "FAILED",
  282. LastError = errorTrunc,
  283. LastDurationMs = durationMs,
  284. LastRunId = runId,
  285. ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
  286. UpdatedAt = now
  287. })
  288. .Where(x => x.Id == rule.Id)
  289. .ExecuteCommandAsync();
  290. }
  291. }
  292. // S8-SCHED-EXEC-1:trigger / recover 抗抖计数兜底,null / <1 / >10 一律按 1,避免非法配置导致永远不建单 / 永远不恢复。
  293. private static int NormalizeAntiflapCount(int raw)
  294. {
  295. if (raw < 1 || raw > 10) return 1;
  296. return raw;
  297. }
  298. /// <summary>
  299. /// 命中时累加 detection_state.consecutive_hit_count;未存在则插入 hitCount=1。
  300. /// 返回当前 state 行(含 Id)以及命中后的 hitCount。
  301. /// 注意:本函数不消费 trigger_count_required;上游决定是否进入 CreateFromHitAsync。
  302. /// </summary>
  303. private async Task<(AdoS8RuleDetectionState? state, int hitCount)> UpsertDetectionStateOnHitAsync(
  304. long tenantId, AdoS8WatchRule rule, S8RuleHit hit)
  305. {
  306. var now = DateTime.Now;
  307. var existing = await _detectionStateRep.AsQueryable()
  308. .Where(x => x.TenantId == tenantId
  309. && x.RuleCode == rule.RuleCode
  310. && x.DedupKey == hit.DedupKey)
  311. .FirstAsync();
  312. if (existing == null)
  313. {
  314. var fresh = new AdoS8RuleDetectionState
  315. {
  316. TenantId = tenantId,
  317. FactoryId = rule.FactoryId,
  318. RuleCode = rule.RuleCode,
  319. DedupKey = hit.DedupKey,
  320. SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? null : hit.SourceObjectType,
  321. SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? null : hit.SourceObjectId,
  322. ConsecutiveHitCount = 1,
  323. ConsecutiveMissCount = 0,
  324. LastSeenAt = now,
  325. LastHitAt = now,
  326. CreatedAt = now,
  327. UpdatedAt = now
  328. };
  329. // BUG-S8-DETECTION-STATE-ACTIVE-EXC-ID-TRIGGER1-001:必须回填 fresh.Id,否则
  330. // trigger=1 首 tick 建单后 UPDATE state SET active_exception_id WHERE id=state.Id
  331. // 命中 0 行(state.Id 为默认 0)。沿用 S8ManualReportService 既定模式。
  332. fresh = await _detectionStateRep.AsInsertable(fresh).ExecuteReturnEntityAsync();
  333. return (fresh, 1);
  334. }
  335. var newHitCount = existing.ConsecutiveHitCount + 1;
  336. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  337. .SetColumns(x => new AdoS8RuleDetectionState
  338. {
  339. ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
  340. ConsecutiveMissCount = 0,
  341. LastSeenAt = now,
  342. LastHitAt = now,
  343. SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? existing.SourceObjectType : hit.SourceObjectType,
  344. SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? existing.SourceObjectId : hit.SourceObjectId,
  345. UpdatedAt = now
  346. })
  347. .Where(x => x.Id == existing.Id)
  348. .ExecuteCommandAsync();
  349. existing.ConsecutiveHitCount = newHitCount;
  350. return (existing, newHitCount);
  351. }
  352. private async Task<long> FindOpenExceptionByDedupKeyAsync(long tenantId, string dedupKey)
  353. {
  354. var ids = await _exceptionRep.AsQueryable()
  355. .Where(x => x.TenantId == tenantId
  356. && !x.IsDeleted
  357. && x.Status != "CLOSED"
  358. && x.DedupKey == dedupKey)
  359. .Select(x => x.Id)
  360. .Take(1)
  361. .ToListAsync();
  362. return ids.Count > 0 ? ids[0] : 0L;
  363. }
  364. /// <summary>
  365. /// R3 OUT_OF_RANGE compat fallback 查找:用 (source_rule_id, related_object_code, status!=CLOSED,
  366. /// dedup_key IS NULL, is_deleted=0) 严格条件定位旧 AlertRule 主链留下的历史记录。
  367. /// </summary>
  368. private async Task<long> FindLegacyOutOfRangeExceptionAsync(long tenantId, long sourceRuleId, string relatedObjectCode)
  369. {
  370. if (string.IsNullOrWhiteSpace(relatedObjectCode)) return 0L;
  371. var ids = await _exceptionRep.AsQueryable()
  372. .Where(x => x.TenantId == tenantId
  373. && !x.IsDeleted
  374. && x.Status != "CLOSED"
  375. && x.DedupKey == null
  376. && x.SourceRuleId == sourceRuleId
  377. && x.RelatedObjectCode == relatedObjectCode)
  378. .Select(x => x.Id)
  379. .Take(1)
  380. .ToListAsync();
  381. return ids.Count > 0 ? ids[0] : 0L;
  382. }
  383. /// <summary>
  384. /// R3 OUT_OF_RANGE compat fallback backfill:把历史记录的 R1 新 6 列(dedup_key/source_rule_code/
  385. /// source_object_type/source_object_id/source_payload/last_detected_at)写入,并刷新 updated_at。
  386. /// </summary>
  387. private async Task BackfillLegacyExceptionAsync(long exceptionId, S8RuleHit hit)
  388. {
  389. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  390. .SetColumns(x => new AdoS8Exception
  391. {
  392. DedupKey = hit.DedupKey,
  393. SourceRuleCode = hit.SourceRuleCode,
  394. SourceObjectType = hit.SourceObjectType,
  395. SourceObjectId = hit.SourceObjectId,
  396. SourcePayload = hit.SourcePayload,
  397. LastDetectedAt = hit.DetectedAt,
  398. UpdatedAt = DateTime.Now
  399. })
  400. .Where(x => x.Id == exceptionId)
  401. .ExecuteCommandAsync();
  402. }
  403. /// <summary>
  404. /// R5 恢复时间最小闭环:对当前 rule 下未关闭、有 dedup_key、recovered_at 仍为 NULL 的异常,
  405. /// 凡不在本轮 hits.dedup_key 集合内的,写入 recovered_at = now、updated_at = now。
  406. /// 仅写这 2 列;不动 status / assignee / verifier / source_payload / last_detected_at;
  407. /// recovered_at 一旦写入,本轮不做复发清空。
  408. /// R6 返回 recoveredIds 供上游决定是否写 NO_HIT 日志,并对每个 recovered exception 写一条 RECOVERED 日志。
  409. /// </summary>
  410. private async Task<List<long>> ReconcileRecoveriesForRuleAsync(
  411. long tenantId, AdoS8WatchRule rule, string ruleType, List<S8RuleHit> hits, string runId)
  412. {
  413. var hitDedupKeys = hits
  414. .Where(h => !string.IsNullOrWhiteSpace(h.DedupKey))
  415. .Select(h => h.DedupKey)
  416. .ToHashSet(StringComparer.Ordinal);
  417. var candidates = await _exceptionRep.AsQueryable()
  418. .Where(x => x.TenantId == tenantId
  419. && !x.IsDeleted
  420. && x.Status != "CLOSED"
  421. && x.SourceRuleCode == rule.RuleCode
  422. && x.DedupKey != null
  423. && x.RecoveredAt == null)
  424. .Select(x => new { x.Id, x.DedupKey, x.SourceObjectType, x.SourceObjectId, x.RelatedObjectCode, x.ConsecutiveMissCount })
  425. .ToListAsync();
  426. if (candidates.Count == 0) return new List<long>();
  427. var now = DateTime.Now;
  428. var recoverRequired = NormalizeAntiflapCount(rule.RecoverCountRequired);
  429. var recoveredIds = new List<long>();
  430. foreach (var c in candidates)
  431. {
  432. if (hitDedupKeys.Contains(c.DedupKey!)) continue;
  433. // S8-SCHED-EXEC-1:恢复抗抖累计。
  434. // 1) 每次未命中:异常 ConsecutiveMissCount += 1,ConsecutiveHitCount 清零;
  435. // 2) miss < recover_count_required:仅累计,不写 recovered_at、不写 RECOVERED;
  436. // 3) miss >= recover_count_required:写 recovered_at、写 RECOVERED 日志。
  437. var newMissCount = c.ConsecutiveMissCount + 1;
  438. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  439. .SetColumns(x => new AdoS8Exception
  440. {
  441. ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
  442. ConsecutiveHitCount = 0,
  443. UpdatedAt = now
  444. })
  445. .Where(x => x.Id == c.Id)
  446. .ExecuteCommandAsync();
  447. // detection_state 同步累计 miss(建单后 state.active_exception_id 仍指向 c.Id)。
  448. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  449. .SetColumns(x => new AdoS8RuleDetectionState
  450. {
  451. ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
  452. ConsecutiveHitCount = 0,
  453. LastSeenAt = now,
  454. LastMissAt = now,
  455. UpdatedAt = now
  456. })
  457. .Where(x => x.TenantId == tenantId
  458. && x.RuleCode == rule.RuleCode
  459. && x.DedupKey == c.DedupKey)
  460. .ExecuteCommandAsync();
  461. if (newMissCount < recoverRequired)
  462. {
  463. _logger.LogInformation(
  464. "antiflap_pending_recovery ruleCode={RuleCode} dedupKey={DedupKey} missCount={Miss} recoverRequired={Required}",
  465. rule.RuleCode, c.DedupKey, newMissCount, recoverRequired);
  466. continue;
  467. }
  468. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  469. .SetColumns(x => new AdoS8Exception
  470. {
  471. RecoveredAt = now,
  472. UpdatedAt = now
  473. })
  474. .Where(x => x.Id == c.Id)
  475. .ExecuteCommandAsync();
  476. recoveredIds.Add(c.Id);
  477. await WriteDetectionLogAsync(new AdoS8DetectionLog
  478. {
  479. TenantId = tenantId, FactoryId = rule.FactoryId,
  480. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  481. SourceObjectType = c.SourceObjectType, SourceObjectId = c.SourceObjectId,
  482. RelatedObjectCode = c.RelatedObjectCode, DedupKey = c.DedupKey,
  483. DetectResult = DetectResultRecovered,
  484. ExceptionId = c.Id,
  485. DetectedAt = now,
  486. PayloadSnapshot = JsonSerializer.Serialize(new { ruleId = rule.Id, ruleCode = rule.RuleCode, reason = "no_longer_hit", missCount = newMissCount, recoverRequired }),
  487. RunId = runId, TriggerSource = DetectionTriggerSource,
  488. Remark = "Rule no longer hit; recovered_at marked"
  489. });
  490. // S8-SYSTEM-RECOVERED-TIMELINE-1:恢复也要在时间线上留痕。
  491. // 此前 recovered_at 与 RECOVERED detection_log 都有了,唯独用户能看见的那条时间线没有 ——
  492. // 于是页面上异常"自己不见了",没人知道是系统判定恢复还是有人做了什么。
  493. await TryWriteSystemRecoveredTimelineAsync(c.Id, now, newMissCount, recoverRequired);
  494. // S8-NOTIFY-WIRE-RECOVERED-1:detection_log 已写入、recovered_at 已落库后挂入恢复通知。
  495. // 通知失败仅 LogWarning,绝不影响恢复状态/检测日志。
  496. await TryDispatchRecoveredLayerNotificationAsync(c.Id);
  497. }
  498. if (recoveredIds.Count > 0)
  499. {
  500. _logger.LogInformation(
  501. "rule_recovered ruleCode={RuleCode} ruleType={RuleType} recoveredCount={Count} recoveredIds={Ids}",
  502. rule.RuleCode, ruleType, recoveredIds.Count, string.Join(",", recoveredIds));
  503. }
  504. return recoveredIds;
  505. }
  506. /// <summary>R6 通用 hit 日志构造(CREATED / REFRESHED 共用)。</summary>
  507. private static AdoS8DetectionLog BuildHitLog(
  508. long tenantId, AdoS8WatchRule rule, string ruleType, S8RuleHit hit,
  509. string detectResult, long exceptionId, string runId) => new()
  510. {
  511. TenantId = tenantId, FactoryId = rule.FactoryId,
  512. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  513. SourceObjectType = hit.SourceObjectType, SourceObjectId = hit.SourceObjectId,
  514. RelatedObjectCode = hit.RelatedObjectCode, DedupKey = hit.DedupKey,
  515. DetectResult = detectResult,
  516. ExceptionId = exceptionId,
  517. DetectedAt = hit.DetectedAt,
  518. PayloadSnapshot = hit.SourcePayload,
  519. RunId = runId,
  520. TriggerSource = DetectionTriggerSource
  521. };
  522. /// <summary>R6 日志写入:失败仅 LogWarning,不阻断主链;不抛异常。</summary>
  523. private async Task WriteDetectionLogAsync(AdoS8DetectionLog log)
  524. {
  525. try
  526. {
  527. await _detectionLogRep.InsertAsync(log);
  528. }
  529. catch (Exception ex)
  530. {
  531. _logger.LogWarning(ex,
  532. "detection_log_write_failed runId={RunId} detectResult={Result} ruleCode={RuleCode} exceptionId={ExceptionId}",
  533. log.RunId, log.DetectResult, log.RuleCode, log.ExceptionId);
  534. }
  535. }
  536. /// <summary>
  537. /// S8-DETECTION-LOG-WRITE-REDUCE-P1-1:REFRESHED 明细限频判定。
  538. /// 查询 cutoff = detectedAt - RefreshedRateLimitMinutes 之后,
  539. /// 是否已有同 (tenant_id, factory_id, rule_code, dedup_key, DetectResult=REFRESHED) 的 detection_log;
  540. /// 命中既有索引 idx_s8_detection_log_dedup_time。查询失败仅 LogWarning 后返回 false,
  541. /// 宁可多写一条 REFRESHED 明细,也不阻断调度主链路;ruleCode / dedupKey 空白时同样返回 false(保留写入)。
  542. /// 业务侧 RefreshDetectionAsync / BackfillLegacyExceptionAsync 不在限频范围。
  543. /// </summary>
  544. private async Task<bool> HasRecentRefreshedDetectionLogAsync(
  545. long tenantId, string ruleCode, string dedupKey, DateTime detectedAt)
  546. {
  547. if (string.IsNullOrWhiteSpace(ruleCode) || string.IsNullOrWhiteSpace(dedupKey))
  548. return false;
  549. var cutoff = detectedAt.AddMinutes(-RefreshedRateLimitMinutes);
  550. try
  551. {
  552. return await _detectionLogRep.AsQueryable()
  553. .Where(x => x.TenantId == tenantId
  554. && x.DedupKey == dedupKey
  555. && x.RuleCode == ruleCode
  556. && x.DetectResult == DetectResultRefreshed
  557. && x.DetectedAt >= cutoff)
  558. .AnyAsync();
  559. }
  560. catch (Exception ex)
  561. {
  562. _logger.LogWarning(ex,
  563. "detection_log_refreshed_ratelimit_check_failed ruleCode={RuleCode} dedupKey={DedupKey}",
  564. ruleCode, dedupKey);
  565. return false;
  566. }
  567. }
  568. /// <summary>
  569. /// S8-SYSTEM-RECOVERED-TIMELINE-1:写一条系统恢复时间线。
  570. ///
  571. /// <para><b>不自动关闭异常</b>:恢复只表示"规则不再命中",是否闭环仍由人判断。
  572. /// 既有语义(recovered_at + RECOVERED detection_log,状态不变)原样保留。</para>
  573. ///
  574. /// <para><b>operator_user_id 写 NULL</b>:这一步没有真人操作者。
  575. /// 绝不伪造 1 / EmployeeId / 某个 SysUserId —— 时间线的操作人列刚在
  576. /// S8-SYSUSER-ONLY-1 里从"混装两套 ID 空间"收拾干净,不能又塞进一个假值。</para>
  577. ///
  578. /// <para><b>幂等判据</b>:不存在 <c>created_at</c> 晚于本异常最近一次命中时间
  579. /// (<c>last_detected_at</c>)的 SYSTEM_RECOVERED 行。
  580. /// <list type="bullet">
  581. /// <item>首次恢复:最近命中之后还没有恢复行 → 写入;</item>
  582. /// <item>调度器重跑:刚写的那行已经晚于 last_detected_at → 跳过,不重复追加;</item>
  583. /// <item>复发后再恢复:<c>RefreshDetectionAsync</c> 会把 last_detected_at 推到
  584. /// 旧恢复行之后 → 写入新的一条,这是<b>真实的新恢复周期</b>,本就该留痕。</item>
  585. /// </list>
  586. /// 只用既有列表达,不新增状态字段。</para>
  587. ///
  588. /// <para>失败仅 LogWarning:时间线写不进去不该把已经判定的恢复状态回滚。</para>
  589. /// </summary>
  590. private async Task TryWriteSystemRecoveredTimelineAsync(long exceptionId, DateTime now, int missCount, int recoverRequired)
  591. {
  592. try
  593. {
  594. var lastDetectedAt = await _exceptionRep.AsQueryable().ClearFilter()
  595. .Where(x => x.Id == exceptionId)
  596. .Select(x => x.LastDetectedAt)
  597. .FirstAsync();
  598. var already = await _exceptionRep.Context.Queryable<AdoS8ExceptionTimeline>()
  599. .Where(t => t.ExceptionId == exceptionId && t.ActionCode == SystemRecoveredActionCode)
  600. .WhereIF(lastDetectedAt != null, t => t.CreatedAt > lastDetectedAt!.Value)
  601. .AnyAsync();
  602. if (already)
  603. {
  604. _logger.LogInformation(
  605. "system_recovered_timeline_skip exceptionId={Id} reason=already_recorded_for_current_cycle", exceptionId);
  606. return;
  607. }
  608. await _exceptionRep.Context.Insertable(new AdoS8ExceptionTimeline
  609. {
  610. ExceptionId = exceptionId,
  611. ActionCode = SystemRecoveredActionCode,
  612. ActionLabel = "系统恢复",
  613. FromStatus = null,
  614. ToStatus = null, // 恢复不改状态机,故不写状态迁移
  615. OperatorUserId = null, // 系统动作无真人操作者,绝不伪造
  616. OperatorName = null,
  617. ActionRemark = $"系统检测到异常已恢复(连续未命中 {missCount} 次,达到恢复阈值 {recoverRequired})。异常未自动关闭,仍需人工确认闭环。",
  618. CreatedAt = now
  619. }).ExecuteCommandAsync();
  620. _logger.LogInformation("system_recovered_timeline_written exceptionId={Id} missCount={Miss}", exceptionId, missCount);
  621. }
  622. catch (Exception ex)
  623. {
  624. _logger.LogWarning(ex, "system_recovered_timeline_failed exceptionId={Id}", exceptionId);
  625. }
  626. }
  627. private static string Truncate(string? s, int max) =>
  628. string.IsNullOrEmpty(s) ? string.Empty : (s.Length <= max ? s : s.Substring(0, max));
  629. private async Task RefreshDetectionAsync(long exceptionId, S8RuleHit hit)
  630. {
  631. // S8-SCHED-EXEC-1:刷新阶段抗抖累计 + 复发清空 recovered_at。
  632. // ConsecutiveHitCount += 1(用 SetColumns 内表达式完成原子自增);ConsecutiveMissCount 归零。
  633. // RecoveredAt 不为 NULL 时(复发)一并清空,保持业务对"再次命中即视为活跃"的预期。
  634. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  635. .SetColumns(x => new AdoS8Exception
  636. {
  637. LastDetectedAt = hit.DetectedAt,
  638. SourcePayload = hit.SourcePayload,
  639. ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
  640. ConsecutiveMissCount = 0,
  641. RecoveredAt = null,
  642. UpdatedAt = DateTime.Now
  643. })
  644. .Where(x => x.Id == exceptionId)
  645. .ExecuteCommandAsync();
  646. }
  647. private static S8WatchCreationResult BuildCreatedResult(AdoS8WatchRule rule, S8RuleHit hit, long exceptionId) =>
  648. new()
  649. {
  650. DedupResult = new S8WatchDedupResult
  651. {
  652. Hit = ToWatchHit(rule, hit),
  653. CanCreate = true,
  654. MatchedExceptionId = null,
  655. Reason = "no_pending"
  656. },
  657. Created = true,
  658. Skipped = false,
  659. CreatedExceptionId = exceptionId,
  660. Reason = "auto_created",
  661. ErrorMessage = null
  662. };
  663. private static S8WatchCreationResult BuildSkippedDuplicate(AdoS8WatchRule rule, S8RuleHit hit, long matchedId) =>
  664. new()
  665. {
  666. DedupResult = new S8WatchDedupResult
  667. {
  668. Hit = ToWatchHit(rule, hit),
  669. CanCreate = false,
  670. MatchedExceptionId = matchedId,
  671. Reason = "duplicate_pending"
  672. },
  673. Created = false,
  674. Skipped = true,
  675. CreatedExceptionId = null,
  676. Reason = "duplicate_pending",
  677. ErrorMessage = null
  678. };
  679. private static S8WatchCreationResult BuildSkipResult(AdoS8WatchRule rule, string reason, string? error, S8RuleHit? hit = null) =>
  680. new()
  681. {
  682. DedupResult = new S8WatchDedupResult
  683. {
  684. Hit = hit != null ? ToWatchHit(rule, hit) : new S8WatchHitResult { SourceRuleId = rule.Id, SourceRuleCode = rule.RuleCode },
  685. CanCreate = false,
  686. MatchedExceptionId = null,
  687. Reason = reason
  688. },
  689. Created = false,
  690. Skipped = true,
  691. CreatedExceptionId = null,
  692. Reason = reason,
  693. ErrorMessage = error
  694. };
  695. private static S8WatchHitResult ToWatchHit(AdoS8WatchRule rule, S8RuleHit hit) => new()
  696. {
  697. SourceRuleId = hit.SourceRuleId == 0 ? rule.Id : hit.SourceRuleId,
  698. SourceRuleCode = string.IsNullOrEmpty(hit.SourceRuleCode) ? rule.RuleCode : hit.SourceRuleCode,
  699. RelatedObjectCode = hit.RelatedObjectCode,
  700. Severity = hit.Severity,
  701. OccurrenceDeptId = hit.OccurrenceDeptId,
  702. ResponsibleDeptId = hit.ResponsibleDeptId,
  703. SourcePayload = hit.SourcePayload
  704. };
  705. // G01-04 首版最小比较符集合:>, >=, <, <=。
  706. // 允许首尾空格;非此集合的一律视为“比较符非法”,由调用方跳过。
  707. private static string NormalizeColumnName(string columnName) =>
  708. columnName.Replace("_", string.Empty, StringComparison.Ordinal).Trim().ToUpperInvariant();
  709. // ============================================================
  710. // S8-SCHED-EXEC-1:DB 驱动调度执行层
  711. // ============================================================
  712. private const int LeaseDurationMinutes = 5;
  713. private const int AutoPauseFailureThreshold = 3;
  714. private const int AutoPauseDurationHours = 1;
  715. private const int DefaultPollIntervalSeconds = 300;
  716. private const int MinPollIntervalSeconds = 60;
  717. private const int MaxPollIntervalSeconds = 86400;
  718. /// <summary>
  719. /// S8-SCHED-EXEC-1:释放过期 lease(lock_until &lt; NOW),不修改 last_status / last_error,仅清空 lock 三件套 + running_started_at。
  720. /// 返回释放的行数。
  721. /// </summary>
  722. public async Task<int> ResetExpiredLeasesAsync(long tenantId)
  723. {
  724. var now = DateTime.Now;
  725. var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  726. .SetColumns(x => new AdoS8WatchRule
  727. {
  728. LockToken = null,
  729. LockedBy = null,
  730. LockUntil = null,
  731. RunningStartedAt = null,
  732. UpdatedAt = now
  733. })
  734. .Where(x => x.TenantId == tenantId
  735. && x.LockUntil != null
  736. && x.LockUntil < now)
  737. .ExecuteCommandAsync();
  738. if (affected > 0)
  739. {
  740. _logger.LogWarning(
  741. "lease_reset tenantId={Tenant} releasedCount={Count}",
  742. tenantId, affected);
  743. }
  744. return affected;
  745. }
  746. /// <summary>
  747. /// S8-SCHED-EXEC-1:到期规则候选 + 乐观 UPDATE 抢锁,返回成功抢到的 lease 列表。
  748. /// 抢锁条件:enabled=1 AND (paused_until IS NULL OR paused_until &lt;= NOW)
  749. /// AND (next_run_at IS NULL OR next_run_at &lt;= NOW)
  750. /// AND (lock_until IS NULL OR lock_until &lt;= NOW)。
  751. /// 抢锁回写:lock_token / locked_by / lock_until = NOW + 5min / running_started_at = NOW / last_run_id = runId。
  752. /// affectedRows == 1 才算抢到;后续 OnRuleCompletedAsync 必须按 lockToken 回写,避免旧进程覆盖新 lease。
  753. /// </summary>
  754. public async Task<List<S8RuleLease>> PickReadyRulesAsync(long tenantId, int batchSize, string lockedBy, string runId)
  755. {
  756. if (batchSize <= 0) batchSize = 16;
  757. var now = DateTime.Now;
  758. var candidates = await _ruleRep.AsQueryable()
  759. .Where(x => x.TenantId == tenantId
  760. && x.Enabled
  761. // S8-STANDARD-DATASET-HARD-CUTOVER-1:取数唯一依赖 dataset_code。
  762. // 保留这条谓词而非依赖 Enable Gate:Gate 只在"切换启用"那一刻生效,
  763. // 而 dataset_code 可能在此之后被清空(历史遗留行、直连改库)。
  764. // 一条没有数据集的规则跑起来只会每 tick 抛一次 dataset_code_missing。
  765. //
  766. // 刻意做成 **SQL 谓词而非内存过滤**:Take(batchSize) 在过滤之前生效,
  767. // 内存过滤会让无效候选占满名额,导致同租户的有效规则被饿死。
  768. && x.DatasetCode != null && x.DatasetCode != ""
  769. && (x.PausedUntil == null || x.PausedUntil <= now)
  770. && (x.NextRunAt == null || x.NextRunAt <= now)
  771. && (x.LockUntil == null || x.LockUntil <= now))
  772. .OrderBy(x => x.NextRunAt, OrderByType.Asc)
  773. .OrderBy(x => x.Id, OrderByType.Asc)
  774. .Take(batchSize)
  775. .Select(x => new { x.Id, x.RuleCode, x.RuleType })
  776. .ToListAsync();
  777. if (candidates.Count == 0) return new();
  778. var leases = new List<S8RuleLease>();
  779. foreach (var c in candidates)
  780. {
  781. var token = Guid.NewGuid().ToString("N");
  782. var lockUntil = DateTime.Now.AddMinutes(LeaseDurationMinutes);
  783. var runningAt = DateTime.Now;
  784. // 乐观 UPDATE:再校验一次条件,affectedRows=1 才算抢到。
  785. var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  786. .SetColumns(x => new AdoS8WatchRule
  787. {
  788. LockToken = token,
  789. LockedBy = lockedBy,
  790. LockUntil = lockUntil,
  791. RunningStartedAt = runningAt,
  792. LastRunId = runId,
  793. UpdatedAt = runningAt
  794. })
  795. .Where(x => x.Id == c.Id
  796. && x.Enabled
  797. // 抢锁的乐观 UPDATE 复查同一条件:候选查询与本次 UPDATE 之间存在时间窗,
  798. // 期间该行的 dataset_code 可能被清空。
  799. && x.DatasetCode != null && x.DatasetCode != ""
  800. && (x.PausedUntil == null || x.PausedUntil <= runningAt)
  801. && (x.NextRunAt == null || x.NextRunAt <= runningAt)
  802. && (x.LockUntil == null || x.LockUntil <= runningAt))
  803. .ExecuteCommandAsync();
  804. if (affected == 1)
  805. {
  806. leases.Add(new S8RuleLease
  807. {
  808. RuleId = c.Id,
  809. RuleCode = c.RuleCode,
  810. RuleType = c.RuleType,
  811. LockToken = token,
  812. LockedBy = lockedBy,
  813. LockUntil = lockUntil,
  814. RunId = runId,
  815. AcquiredAt = runningAt
  816. });
  817. }
  818. }
  819. return leases;
  820. }
  821. /// <summary>
  822. /// S8-SCHED-EXEC-1:执行单条已抢锁规则的 evaluator → 抗抖去重 → 建单/刷新 → 恢复 reconcile。
  823. /// 不释放 lease(OnRuleCompletedAsync 负责);evaluator 抛异常时 Result.Success=false 并保留 ErrorMessage。
  824. /// </summary>
  825. public async Task<S8RuleRunResult> RunSingleRuleAsync(long tenantId, S8RuleLease lease)
  826. {
  827. // S8-P0-1-SCHEDULER-TRUSTED-SCOPE-1:执行入口必须按本次 tick 的可信作用域绑行。
  828. // 原实现只按 lease.RuleId 装载,不校验 rule 归属;一旦 lease 来源被改写 / 未来新增调用方,
  829. // 就会在 A 租户的 scope 上下文里执行 B 租户的规则,而下游全部以传入的 tenantId/factoryId 落库。
  830. // 这里同时做「谓词绑行」与「归属复核」两层,任一不符即 fail-fast,绝不静默降级。
  831. var rule = await _ruleRep.AsQueryable()
  832. .Where(x => x.Id == lease.RuleId && x.TenantId == tenantId)
  833. .FirstAsync();
  834. if (rule == null)
  835. {
  836. return new S8RuleRunResult { Success = false, ErrorMessage = "rule_not_found", Stats = new() };
  837. }
  838. if (rule.TenantId != tenantId)
  839. {
  840. _logger.LogError(
  841. "rule_scope_mismatch ruleId={RuleId} ruleTenant={RuleTenant} scopeTenant={ScopeTenant}",
  842. rule.Id, rule.TenantId, tenantId);
  843. return new S8RuleRunResult { Success = false, ErrorMessage = "rule_scope_mismatch", Stats = new() };
  844. }
  845. // S8-RULE-GOVERNANCE-BATCH1:分派的真源是**代码定义**,不是 DB 上的 rule_type 列。
  846. // 那一列现在只是 provisioning 维护的投影;若有人手工改它,规则的执行方式不应随之改变。
  847. // 同时这里是「没有代码定义的规则不得进入执行链」的把关点 —— 在 Create API 仍存在的
  848. // 过渡期(Batch 3 才退役),它保证业务即使造出任意 rule_code 也调不动调度器。
  849. var definition = _ruleCatalog.TryGet(rule.RuleCode);
  850. if (definition == null)
  851. {
  852. _logger.LogWarning(
  853. "rule_definition_not_found ruleId={RuleId} ruleCode={RuleCode} tenant={Tenant}",
  854. rule.Id, rule.RuleCode, tenantId);
  855. return new S8RuleRunResult { Success = false, ErrorMessage = S8RuleCatalog.ReasonNotFound, Stats = new() };
  856. }
  857. // S8-RULE-READINESS-1:第三条真实执行入口。**必须在 evaluator 之前** fail-fast。
  858. //
  859. // 为什么调度器也要查:启用那一刻合法,不代表现在仍合法 —— 默认部门可能被停用、
  860. // 删除,或主数据被改到别的租户。放行到下游的后果不是"报个错",而是
  861. // tick_rule_done status=SUCCESS hits=28 created=0 failed=28
  862. // 每一层都成功、异常一条没有,页面上完全看不出原因(Batch 7 实测)。
  863. //
  864. // 失败语义:本次 run 直接判失败并带上 reason,由 OnRuleCompletedAsync 落到
  865. // last_status=FAILED / last_error;**不自动停用规则**,配好部门后自然恢复。
  866. var readiness = await _readinessGate.CheckAsync(
  867. S8EffectiveRule.Resolve(rule, definition), tenantId);
  868. if (!readiness.Ok)
  869. {
  870. _logger.LogWarning(
  871. "rule_not_ready ruleId={RuleId} ruleCode={RuleCode} tenant={Tenant} reason={Reason} message={Message}",
  872. rule.Id, rule.RuleCode, tenantId, readiness.ReasonCode, readiness.Message);
  873. return new S8RuleRunResult
  874. {
  875. Success = false,
  876. ErrorMessage = $"{readiness.ReasonCode}: {readiness.Message}",
  877. Stats = new()
  878. };
  879. }
  880. var ruleType = definition.RuleType;
  881. IS8RuleEvaluator? evaluator = ruleType switch
  882. {
  883. S8TimeoutRuleEvaluator.RuleTypeCode => _timeoutEvaluator,
  884. S8ShortageRuleEvaluator.RuleTypeCode => _shortageEvaluator,
  885. S8OutOfRangeRuleEvaluator.RuleTypeCode => _outOfRangeEvaluator,
  886. _ => null
  887. };
  888. if (evaluator == null)
  889. {
  890. return new S8RuleRunResult { Success = false, ErrorMessage = $"unsupported_rule_type:{ruleType}", Stats = new() };
  891. }
  892. try
  893. {
  894. var results = await ProcessSingleRuleAsync(tenantId, rule, ruleType, evaluator, lease.RunId);
  895. var stats = new S8RuleRunStats
  896. {
  897. Hits = results.Count,
  898. Created = results.Count(r => r.Created),
  899. Refreshed = results.Count(r => r.Reason == "duplicate_pending"),
  900. Pending = results.Count(r => r.Reason == "antiflap_pending_hit"),
  901. Failed = results.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
  902. };
  903. return new S8RuleRunResult { Success = true, Stats = stats };
  904. }
  905. catch (Exception ex)
  906. {
  907. return new S8RuleRunResult
  908. {
  909. Success = false,
  910. ErrorMessage = ex.Message,
  911. Stats = new()
  912. };
  913. }
  914. }
  915. /// <summary>
  916. /// S8-SCHED-EXEC-1:单规则处理(evaluator → reconcile → hit 循环)。
  917. /// 与 ProcessRulesByTypeAsync 内单规则循环体语义一致;此处抽出便于新调度路径直接调用单条 rule。
  918. /// </summary>
  919. private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync(
  920. long tenantId, AdoS8WatchRule rule, string ruleType,
  921. IS8RuleEvaluator evaluator, string runId)
  922. {
  923. var results = new List<S8WatchCreationResult>();
  924. List<S8RuleHit> hits;
  925. try
  926. {
  927. hits = await evaluator.EvaluateAsync(tenantId, rule);
  928. }
  929. catch (Exception ex)
  930. {
  931. var failureReason = ex is S8RuleEvaluatorException sre ? sre.Reason : ex.GetType().Name;
  932. await WriteDetectionLogAsync(new AdoS8DetectionLog
  933. {
  934. TenantId = tenantId, FactoryId = rule.FactoryId,
  935. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  936. SourceObjectType = rule.SourceObjectType,
  937. DetectResult = DetectResultEvaluateFailed,
  938. DetectedAt = DateTime.Now,
  939. FailureReason = failureReason,
  940. FailureMessage = Truncate(ex.Message, 1000),
  941. RunId = runId, TriggerSource = DetectionTriggerSource
  942. });
  943. results.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
  944. throw;
  945. }
  946. List<long> recoveredIds;
  947. try
  948. {
  949. recoveredIds = await ReconcileRecoveriesForRuleAsync(tenantId, rule, ruleType, hits, runId);
  950. }
  951. catch (Exception ex)
  952. {
  953. _logger.LogWarning(ex, "recovery_reconcile_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  954. recoveredIds = new();
  955. }
  956. // S8-DETECTION-LOG-WRITE-REDUCE-P1-1:NO_HIT 不再写 detection_log 明细。
  957. // 规则级"被定期评估"信号由 ado_s8_watch_rule.LastRunAt / LastStatus / LastRunId / LastDurationMs
  958. // 在 OnRuleCompletedAsync / ApplyRunOnceCompletionAsync 中承接,无需 detection_log 冗余记录。
  959. // DetectResultNoHit 常量保留(历史数据反查 / 未来如需恢复明细写入)。
  960. foreach (var hit in hits)
  961. {
  962. if (string.IsNullOrWhiteSpace(hit.DedupKey))
  963. {
  964. results.Add(BuildSkipResult(rule, "missing_dedup_key", null, hit));
  965. continue;
  966. }
  967. long matchedId;
  968. try
  969. {
  970. matchedId = await FindOpenExceptionByDedupKeyAsync(tenantId, hit.DedupKey);
  971. }
  972. catch (Exception ex)
  973. {
  974. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  975. continue;
  976. }
  977. if (matchedId > 0)
  978. {
  979. try
  980. {
  981. await RefreshDetectionAsync(matchedId, hit);
  982. if (!await HasRecentRefreshedDetectionLogAsync(tenantId, rule.RuleCode, hit.DedupKey, hit.DetectedAt))
  983. {
  984. await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultRefreshed, matchedId, runId));
  985. }
  986. results.Add(BuildSkippedDuplicate(rule, hit, matchedId));
  987. }
  988. catch (Exception ex)
  989. {
  990. results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
  991. }
  992. continue;
  993. }
  994. if (string.Equals(ruleType, S8OutOfRangeRuleEvaluator.RuleTypeCode, StringComparison.OrdinalIgnoreCase))
  995. {
  996. long compatId;
  997. try
  998. {
  999. compatId = await FindLegacyOutOfRangeExceptionAsync(tenantId, rule.Id, hit.RelatedObjectCode);
  1000. }
  1001. catch (Exception ex)
  1002. {
  1003. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  1004. continue;
  1005. }
  1006. if (compatId > 0)
  1007. {
  1008. try
  1009. {
  1010. await BackfillLegacyExceptionAsync(compatId, hit);
  1011. if (!await HasRecentRefreshedDetectionLogAsync(tenantId, rule.RuleCode, hit.DedupKey, hit.DetectedAt))
  1012. {
  1013. await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultRefreshed, compatId, runId));
  1014. }
  1015. results.Add(BuildSkippedDuplicate(rule, hit, compatId));
  1016. }
  1017. catch (Exception ex)
  1018. {
  1019. results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
  1020. }
  1021. continue;
  1022. }
  1023. }
  1024. bool typeExists;
  1025. try
  1026. {
  1027. typeExists = await _exceptionTypeRep.AsQueryable()
  1028. // S8-TENANT-ONLY-BATCH5:去掉工厂条件。这是一次**存在性**检查,
  1029. // 少一个条件只会更宽松:factory_id=0 的供给规则也能看到本租户的工厂级异常类型行,
  1030. // 否则它只能匹配到平台默认行、找不到就整条规则 exception_type_missing。
  1031. // 注意 t.TenantId == 0 是平台默认哨兵(S8ConfigScope),不是工厂语义,保留。
  1032. .Where(t => t.TypeCode == hit.ExceptionTypeCode
  1033. && (t.TenantId == 0 || t.TenantId == tenantId)
  1034. && t.Enabled)
  1035. .AnyAsync();
  1036. }
  1037. catch (Exception ex)
  1038. {
  1039. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  1040. continue;
  1041. }
  1042. if (!typeExists)
  1043. {
  1044. results.Add(BuildSkipResult(rule, "exception_type_missing", null, hit));
  1045. continue;
  1046. }
  1047. int hitCount;
  1048. AdoS8RuleDetectionState? state;
  1049. try
  1050. {
  1051. (state, hitCount) = await UpsertDetectionStateOnHitAsync(tenantId, rule, hit);
  1052. }
  1053. catch (Exception ex)
  1054. {
  1055. results.Add(BuildSkipResult(rule, "antiflap_failed", ex.Message, hit));
  1056. continue;
  1057. }
  1058. var triggerRequired = NormalizeAntiflapCount(rule.TriggerCountRequired);
  1059. if (hitCount < triggerRequired)
  1060. {
  1061. _logger.LogInformation(
  1062. "antiflap_pending_hit ruleCode={RuleCode} dedupKey={DedupKey} hitCount={HitCount} trigger={Trigger}",
  1063. rule.RuleCode, hit.DedupKey, hitCount, triggerRequired);
  1064. results.Add(BuildSkipResult(rule, "antiflap_pending_hit", null, hit));
  1065. continue;
  1066. }
  1067. try
  1068. {
  1069. var entity = await _reportDispatcher.CreateFromHitAsync(tenantId, hit);
  1070. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  1071. .SetColumns(x => new AdoS8Exception
  1072. {
  1073. ConsecutiveHitCount = hitCount,
  1074. ConsecutiveMissCount = 0,
  1075. UpdatedAt = DateTime.Now
  1076. })
  1077. .Where(x => x.Id == entity.Id)
  1078. .ExecuteCommandAsync();
  1079. if (state != null)
  1080. {
  1081. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  1082. .SetColumns(x => new AdoS8RuleDetectionState
  1083. {
  1084. ActiveExceptionId = entity.Id,
  1085. UpdatedAt = DateTime.Now
  1086. })
  1087. .Where(x => x.Id == state.Id)
  1088. .ExecuteCommandAsync();
  1089. }
  1090. await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultCreated, entity.Id, runId));
  1091. await TryDispatchLayerNotificationAsync(entity);
  1092. results.Add(BuildCreatedResult(rule, hit, entity.Id));
  1093. }
  1094. catch (Exception ex)
  1095. {
  1096. results.Add(BuildSkipResult(rule, "create_failed", ex.Message, hit));
  1097. }
  1098. }
  1099. return results;
  1100. }
  1101. /// <summary>
  1102. /// S8-SCHED-EXEC-1:lease 执行完成回写。
  1103. /// 必须 WHERE id = lease.RuleId AND lock_token = lease.LockToken;affectedRows = 0 视为 lease 丢失,记录 Warning,不覆盖状态。
  1104. /// 失败 ≥ 阈值(默认 3)写 paused_until = NOW + 1h。
  1105. /// </summary>
  1106. public async Task OnRuleCompletedAsync(long tenantId, S8RuleLease lease, S8RuleRunResult result, int durationMs)
  1107. {
  1108. var rule = await _ruleRep.AsQueryable().Where(x => x.Id == lease.RuleId).FirstAsync();
  1109. if (rule == null)
  1110. {
  1111. _logger.LogWarning("lease_complete_rule_missing ruleId={RuleId}", lease.RuleId);
  1112. return;
  1113. }
  1114. var now = DateTime.Now;
  1115. var effectiveInterval = NormalizePollInterval(rule.PollIntervalSeconds);
  1116. var nextRunAt = now.AddSeconds(effectiveInterval);
  1117. int affected;
  1118. if (result.Success)
  1119. {
  1120. affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  1121. .SetColumns(x => new AdoS8WatchRule
  1122. {
  1123. LastRunAt = now,
  1124. NextRunAt = nextRunAt,
  1125. LastStatus = "SUCCESS",
  1126. LastError = null,
  1127. LastDurationMs = durationMs,
  1128. ConsecutiveFailureCount = 0,
  1129. LockToken = null,
  1130. LockedBy = null,
  1131. LockUntil = null,
  1132. RunningStartedAt = null,
  1133. UpdatedAt = now
  1134. })
  1135. .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
  1136. .ExecuteCommandAsync();
  1137. }
  1138. else
  1139. {
  1140. var errorTrunc = Truncate(result.ErrorMessage, 500);
  1141. var newFailures = rule.ConsecutiveFailureCount + 1;
  1142. DateTime? pausedUntil = rule.PausedUntil;
  1143. string? pauseReason = rule.PauseReason;
  1144. if (newFailures >= AutoPauseFailureThreshold)
  1145. {
  1146. pausedUntil = now.AddHours(AutoPauseDurationHours);
  1147. pauseReason = Truncate($"AUTO_PAUSED_AFTER_{AutoPauseFailureThreshold}_FAILURES: {errorTrunc}", 64);
  1148. }
  1149. affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  1150. .SetColumns(x => new AdoS8WatchRule
  1151. {
  1152. LastRunAt = now,
  1153. NextRunAt = nextRunAt,
  1154. LastStatus = "FAILED",
  1155. LastError = errorTrunc,
  1156. LastDurationMs = durationMs,
  1157. ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
  1158. PausedUntil = pausedUntil,
  1159. PauseReason = pauseReason,
  1160. LockToken = null,
  1161. LockedBy = null,
  1162. LockUntil = null,
  1163. RunningStartedAt = null,
  1164. UpdatedAt = now
  1165. })
  1166. .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
  1167. .ExecuteCommandAsync();
  1168. }
  1169. if (affected == 0)
  1170. {
  1171. _logger.LogWarning(
  1172. "lease_lost_on_complete ruleId={RuleId} ruleCode={RuleCode} lockToken={LockToken}",
  1173. lease.RuleId, lease.RuleCode, lease.LockToken);
  1174. }
  1175. }
  1176. /// <summary>
  1177. /// S8-SCHED-EXEC-1:单 tick 完整流程。Job / debug 调度入口。
  1178. /// 1) ResetExpiredLeasesAsync
  1179. /// 2) PickReadyRulesAsync(batchSize)
  1180. /// 3) 每条 rule 独立 try/catch 调用 RunSingleRuleAsync + OnRuleCompletedAsync
  1181. /// 单条规则失败不影响其他规则;整 tick 不抛异常。
  1182. /// </summary>
  1183. public async Task<S8DispatchTickResult> RunDispatchTickAsync(long tenantId, int batchSize, string lockedBy)
  1184. {
  1185. var tickId = Guid.NewGuid().ToString("N").Substring(0, 8);
  1186. var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
  1187. var summary = new S8DispatchTickResult { TickId = tickId, RunId = runId };
  1188. try
  1189. {
  1190. summary.LeaseReleased = await ResetExpiredLeasesAsync(tenantId);
  1191. }
  1192. catch (Exception ex)
  1193. {
  1194. _logger.LogError(ex, "tick_reset_lease_failed tickId={TickId}", tickId);
  1195. }
  1196. List<S8RuleLease> leases;
  1197. try
  1198. {
  1199. leases = await PickReadyRulesAsync(tenantId, batchSize, lockedBy, runId);
  1200. }
  1201. catch (Exception ex)
  1202. {
  1203. _logger.LogError(ex, "tick_pick_failed tickId={TickId}", tickId);
  1204. return summary;
  1205. }
  1206. summary.Picked = leases.Count;
  1207. foreach (var lease in leases)
  1208. {
  1209. var sw = System.Diagnostics.Stopwatch.StartNew();
  1210. S8RuleRunResult runResult;
  1211. try
  1212. {
  1213. runResult = await RunSingleRuleAsync(tenantId, lease);
  1214. }
  1215. catch (Exception ex)
  1216. {
  1217. runResult = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
  1218. }
  1219. sw.Stop();
  1220. try
  1221. {
  1222. await OnRuleCompletedAsync(tenantId, lease, runResult, (int)sw.ElapsedMilliseconds);
  1223. }
  1224. catch (Exception ex)
  1225. {
  1226. _logger.LogError(ex, "tick_complete_failed tickId={TickId} ruleId={RuleId} ruleCode={RuleCode}", tickId, lease.RuleId, lease.RuleCode);
  1227. }
  1228. _logger.LogInformation(
  1229. "tick_rule_done tickId={TickId} runId={RunId} ruleId={RuleId} ruleCode={RuleCode} status={Status} durationMs={Dur} hits={Hits} created={Created} refreshed={Refreshed} pending={Pending} failed={Failed} error={Error}",
  1230. tickId, runId, lease.RuleId, lease.RuleCode,
  1231. runResult.Success ? "SUCCESS" : "FAILED",
  1232. sw.ElapsedMilliseconds,
  1233. runResult.Stats.Hits, runResult.Stats.Created, runResult.Stats.Refreshed, runResult.Stats.Pending, runResult.Stats.Failed,
  1234. runResult.ErrorMessage);
  1235. if (runResult.Success)
  1236. {
  1237. summary.Success++;
  1238. summary.Created += runResult.Stats.Created;
  1239. summary.Refreshed += runResult.Stats.Refreshed;
  1240. summary.Pending += runResult.Stats.Pending;
  1241. summary.PerRuleFailed += runResult.Stats.Failed;
  1242. }
  1243. else
  1244. {
  1245. summary.Failed++;
  1246. }
  1247. }
  1248. return summary;
  1249. }
  1250. private static int NormalizePollInterval(int raw)
  1251. {
  1252. if (raw < MinPollIntervalSeconds || raw > MaxPollIntervalSeconds) return DefaultPollIntervalSeconds;
  1253. return raw;
  1254. }
  1255. /// <summary>
  1256. /// S8-NOTIFY-WIRE-WATCH-1:异常自动建单成功后,非破坏性挂入通知分层路由。
  1257. /// 全程异常隔离:任何异常仅 LogWarning,不抛回主流程,不影响 detection_log / 事务 /
  1258. /// 异常状态机;该方法独立于 CreateFromWatchAsync / CreateFromHitAsync 的事务边界
  1259. /// (两者已 commit 后才返回 entity,故在此调用安全)。
  1260. /// sceneCode:先从 entity.SceneCode;空则 fallback 到 "S8_DEMO_DEFAULT"(demo 路径)。
  1261. /// severity:直接取 entity.Severity(CreateFromWatchAsync/CreateFromHitAsync 已保证非空)。
  1262. /// </summary>
  1263. private async Task TryDispatchLayerNotificationAsync(AdoS8Exception entity)
  1264. {
  1265. if (entity == null || entity.Id <= 0) return;
  1266. try
  1267. {
  1268. var sceneCode = string.IsNullOrWhiteSpace(entity.SceneCode) ? "S8_DEMO_DEFAULT" : entity.SceneCode;
  1269. // S8-SEVERITY-FOLLOW-SERIOUS-STANDARDIZE-EXEC-1:写入前 Normalize,落 FOLLOW/SERIOUS。
  1270. var severity = S8SeverityCode.Normalize(entity.Severity);
  1271. // S8-DEMO-IMPACT-SORT-NOTICE-1:附带 30 天影响统计(重复次数 / 累计损失 / 建议关注级别)。
  1272. // 计算失败不阻断通知主链路;snap==null 时 BuildNotification 仅写基础 Context。
  1273. ExceptionImpactSnapshot? snap = null;
  1274. try
  1275. {
  1276. snap = await _impactMetricsService.ComputeOneAsync(entity);
  1277. }
  1278. catch (Exception ex)
  1279. {
  1280. _logger.LogWarning(ex, "notify_impact_compute_throw exceptionId={ExceptionId}", entity.Id);
  1281. }
  1282. var baseContent =
  1283. $"异常 {entity.ExceptionCode}:{entity.Title}(场景 {sceneCode},严重度 {severity}" +
  1284. (string.IsNullOrWhiteSpace(entity.SourceRuleCode) ? "" : $",规则 {entity.SourceRuleCode}") + ")";
  1285. var content = snap == null
  1286. ? baseContent
  1287. : baseContent
  1288. + Environment.NewLine
  1289. + $"过去30天同类异常发生 {snap.RepeatCount30d} 次,已关闭累计损失 {snap.CumulativeLossHours30d.ToString("0.#")} 小时,建议关注级别:{snap.SuggestedAttentionLabel}";
  1290. await _notificationLayerResolver.DispatchByLayerAsync(new S8NotificationLayerResolver.DispatchByLayerInput
  1291. {
  1292. TenantId = entity.TenantId,
  1293. ExceptionId = entity.Id,
  1294. ExceptionNo = entity.ExceptionCode,
  1295. SceneCode = sceneCode,
  1296. Severity = severity,
  1297. EventCode = S8NotifyEventCode.ExceptionCreated,
  1298. ExceptionRef = entity,
  1299. Title = entity.Title ?? string.Empty,
  1300. Content = content,
  1301. Status = entity.Status,
  1302. SourceRuleCode = entity.SourceRuleCode,
  1303. JumpUrl = $"/aidop/s8/exceptions/{entity.Id}",
  1304. RepeatCount30d = snap?.RepeatCount30d,
  1305. CumulativeLossHours30d = snap?.CumulativeLossHours30d,
  1306. SuggestedAttentionLevel = snap?.SuggestedAttentionLevel,
  1307. SuggestedAttentionLabel = snap?.SuggestedAttentionLabel,
  1308. ImpactReason = snap?.ImpactReason,
  1309. });
  1310. }
  1311. catch (Exception ex)
  1312. {
  1313. _logger.LogWarning(ex, "notify_dispatch_throw exceptionId={ExceptionId}", entity.Id);
  1314. }
  1315. }
  1316. /// <summary>
  1317. /// S8-NOTIFY-WIRE-RECOVERED-1:异常恢复(recovered_at 已写入、RECOVERED detection_log 已落库)后,
  1318. /// 非破坏性挂入分层通知。全程异常隔离:任何异常仅 LogWarning,不抛回主流程,不影响 detection_log /
  1319. /// 状态机;call site 已在事务边界外(recovered 路径无事务)。
  1320. /// 重新读取 entity 拿场景/严重度/编号/状态/规则代码(恢复事件相对低频,1 次额外读可接受)。
  1321. /// S8-DEMO-IMPACT-SORT-NOTICE-1:恢复事件**不**追加影响统计字段,保持原有恢复语义不变。
  1322. /// </summary>
  1323. private async Task TryDispatchRecoveredLayerNotificationAsync(long exceptionId)
  1324. {
  1325. if (exceptionId <= 0) return;
  1326. try
  1327. {
  1328. var entity = await _exceptionRep.GetByIdAsync(exceptionId);
  1329. if (entity == null)
  1330. {
  1331. _logger.LogWarning("notify_recovered_dispatch_entity_missing exceptionId={ExceptionId}", exceptionId);
  1332. return;
  1333. }
  1334. var sceneCode = string.IsNullOrWhiteSpace(entity.SceneCode) ? "S8_DEMO_DEFAULT" : entity.SceneCode;
  1335. // S8-SEVERITY-FOLLOW-SERIOUS-STANDARDIZE-EXEC-1:写入前 Normalize,落 FOLLOW/SERIOUS。
  1336. var severity = S8SeverityCode.Normalize(entity.Severity);
  1337. var title = $"【已恢复】{entity.ExceptionCode}";
  1338. var content =
  1339. $"异常 {entity.ExceptionCode} 已恢复,场景 {sceneCode},严重度 {severity}" +
  1340. (string.IsNullOrWhiteSpace(entity.SourceRuleCode) ? "" : $",规则 {entity.SourceRuleCode}");
  1341. await _notificationLayerResolver.DispatchByLayerAsync(new S8NotificationLayerResolver.DispatchByLayerInput
  1342. {
  1343. TenantId = entity.TenantId,
  1344. ExceptionId = entity.Id,
  1345. ExceptionNo = entity.ExceptionCode,
  1346. SceneCode = sceneCode,
  1347. Severity = severity,
  1348. EventCode = S8NotifyEventCode.ExceptionRecovered,
  1349. ExceptionRef = entity,
  1350. Title = title,
  1351. Content = content,
  1352. Status = entity.Status,
  1353. SourceRuleCode = entity.SourceRuleCode,
  1354. JumpUrl = $"/aidop/s8/exceptions/{entity.Id}",
  1355. Recovered = true,
  1356. });
  1357. }
  1358. catch (Exception ex)
  1359. {
  1360. _logger.LogWarning(ex, "notify_recovered_dispatch_throw exceptionId={ExceptionId}", exceptionId);
  1361. }
  1362. }
  1363. }
  1364. /// <summary>
  1365. /// G01-05 去重结果对象。仅服务 G01-06 建单前拦截,由 CanCreate 单决策位决定是否建单。
  1366. /// 只服务首版唯一场景 S2(迁移后由 S2S6_PRODUCTION 切到单模块 S2)+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1367. /// 不预留多 trigger_type / 平台化去重扩展结构。
  1368. /// Reason 值域:no_pending / duplicate_pending / missing_dedup_key / query_failed。
  1369. /// </summary>
  1370. public sealed class S8WatchDedupResult
  1371. {
  1372. public S8WatchHitResult Hit { get; set; } = new();
  1373. public bool CanCreate { get; set; }
  1374. public long? MatchedExceptionId { get; set; }
  1375. public string Reason { get; set; } = string.Empty;
  1376. }
  1377. /// <summary>
  1378. /// G01-06 建单结果对象。仅服务 G-01 首版主线验收,由 Created / Skipped 两位决定结局。
  1379. /// 只服务首版唯一场景 S2(迁移后由 S2S6_PRODUCTION 切到单模块 S2)+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1380. /// 不预留多 trigger_type / 平台化工单扩展结构。
  1381. /// Reason 值域:auto_created / create_failed / 透传自 DedupResult.Reason。
  1382. /// </summary>
  1383. public sealed class S8WatchCreationResult
  1384. {
  1385. public S8WatchDedupResult DedupResult { get; set; } = new();
  1386. public bool Created { get; set; }
  1387. public bool Skipped { get; set; }
  1388. public long? CreatedExceptionId { get; set; }
  1389. public string Reason { get; set; } = string.Empty;
  1390. public string? ErrorMessage { get; set; }
  1391. }
  1392. /// <summary>
  1393. /// G01-04 命中结果对象。承载 G01-05 去重与 G01-06 建单所需最小追溯字段,
  1394. /// 仅服务首版唯一场景 S2(迁移后由 S2S6_PRODUCTION 切到单模块 S2)+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1395. /// 不预留多 trigger_type / 多场景 / 平台化扩展结构。
  1396. /// </summary>
  1397. public sealed class S8WatchHitResult
  1398. {
  1399. public long SourceRuleId { get; set; }
  1400. public string SourceRuleCode { get; set; } = string.Empty;
  1401. public string RelatedObjectCode { get; set; } = string.Empty;
  1402. public decimal CurrentValue { get; set; }
  1403. public decimal ThresholdValue { get; set; }
  1404. public string TriggerCondition { get; set; } = string.Empty;
  1405. public string Severity { get; set; } = string.Empty;
  1406. public long? OccurrenceDeptId { get; set; }
  1407. public long? ResponsibleDeptId { get; set; }
  1408. public string SourcePayload { get; set; } = string.Empty;
  1409. }
  1410. /// <summary>S8-SCHED-EXEC-1:lease 抢占成功后传递的最小标识对象。</summary>
  1411. public sealed class S8RuleLease
  1412. {
  1413. public long RuleId { get; set; }
  1414. public string RuleCode { get; set; } = string.Empty;
  1415. public string? RuleType { get; set; }
  1416. public string LockToken { get; set; } = string.Empty;
  1417. public string LockedBy { get; set; } = string.Empty;
  1418. public DateTime LockUntil { get; set; }
  1419. public string RunId { get; set; } = string.Empty;
  1420. public DateTime AcquiredAt { get; set; }
  1421. }
  1422. /// <summary>S8-SCHED-EXEC-1:单条规则执行结果,OnRuleCompletedAsync 据此更新状态。</summary>
  1423. public sealed class S8RuleRunResult
  1424. {
  1425. public bool Success { get; set; }
  1426. public string? ErrorMessage { get; set; }
  1427. public S8RuleRunStats Stats { get; set; } = new();
  1428. }
  1429. public sealed class S8RuleRunStats
  1430. {
  1431. public int Hits { get; set; }
  1432. public int Created { get; set; }
  1433. public int Refreshed { get; set; }
  1434. public int Pending { get; set; }
  1435. public int Failed { get; set; }
  1436. }
  1437. /// <summary>S8-SCHED-EXEC-1:单 tick 调度结果聚合。</summary>
  1438. public sealed class S8DispatchTickResult
  1439. {
  1440. public string TickId { get; set; } = string.Empty;
  1441. public string RunId { get; set; } = string.Empty;
  1442. public int LeaseReleased { get; set; }
  1443. public int Picked { get; set; }
  1444. public int Success { get; set; }
  1445. public int Failed { get; set; }
  1446. public int Created { get; set; }
  1447. public int Refreshed { get; set; }
  1448. public int Pending { get; set; }
  1449. public int PerRuleFailed { get; set; }
  1450. }