S8WatchSchedulerService.cs 75 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733
  1. using Admin.NET.Plugin.AiDOP.Entity.S8;
  2. using Admin.NET.Plugin.AiDOP.Infrastructure.S8;
  3. using Admin.NET.Plugin.AiDOP.Service.S8.Rules;
  4. using Microsoft.Extensions.Logging;
  5. using SqlSugar;
  6. using System.Data;
  7. using System.Globalization;
  8. using System.Text.Json;
  9. namespace Admin.NET.Plugin.AiDOP.Service.S8;
  10. /// <summary>
  11. /// 监视规则轮询调度服务(首轮存根)。
  12. /// 后续接入 Admin.NET 定时任务机制后,由调度器周期调用 <see cref="RunOnceAsync"/>,
  13. /// 按各规则的 PollIntervalSeconds 逐条评估并生成异常记录。
  14. /// </summary>
  15. public class S8WatchSchedulerService : ITransient
  16. {
  17. private readonly SqlSugarRepository<AdoS8WatchRule> _ruleRep;
  18. private readonly SqlSugarRepository<AdoS8AlertRule> _alertRuleRep;
  19. private readonly SqlSugarRepository<AdoS8DataSource> _dataSourceRep;
  20. private readonly SqlSugarRepository<AdoS8Exception> _exceptionRep;
  21. private readonly SqlSugarRepository<AdoS8ExceptionType> _exceptionTypeRep;
  22. private readonly S8NotificationService _notificationService;
  23. private readonly S8ManualReportService _manualReportService;
  24. private readonly S8TimeoutRuleEvaluator _timeoutEvaluator;
  25. private readonly S8ShortageRuleEvaluator _shortageEvaluator;
  26. private readonly S8OutOfRangeRuleEvaluator _outOfRangeEvaluator;
  27. private readonly ILogger<S8WatchSchedulerService> _logger;
  28. private readonly SqlSugarRepository<AdoS8DetectionLog> _detectionLogRep;
  29. private readonly SqlSugarRepository<AdoS8RuleDetectionState> _detectionStateRep;
  30. private const string DetectionTriggerSource = "WATCH_SCHEDULER";
  31. private const string DetectResultCreated = "CREATED";
  32. private const string DetectResultRefreshed = "REFRESHED";
  33. private const string DetectResultRecovered = "RECOVERED";
  34. private const string DetectResultNoHit = "NO_HIT";
  35. private const string DetectResultEvaluateFailed = "EVALUATE_FAILED";
  36. private const string DefaultTriggerType = "VALUE_DEVIATION";
  37. private const string SqlDataSourceType = "SQL";
  38. // G01-05 未闭环状态集合:复用自 S8ExceptionService 当前 pendingStatuses 事实口径
  39. // (见 S8ExceptionService.GetPagedAsync 中 pendingStatuses 的定义,两处必须保持一致)。
  40. // 这不是“自定义未闭环集合”;若现有口径调整,两处需同步修改。
  41. private static readonly string[] UnclosedExceptionStatuses =
  42. { "NEW", "ASSIGNED", "IN_PROGRESS", "PENDING_VERIFICATION" };
  43. public S8WatchSchedulerService(
  44. SqlSugarRepository<AdoS8WatchRule> ruleRep,
  45. SqlSugarRepository<AdoS8AlertRule> alertRuleRep,
  46. SqlSugarRepository<AdoS8DataSource> dataSourceRep,
  47. SqlSugarRepository<AdoS8Exception> exceptionRep,
  48. SqlSugarRepository<AdoS8ExceptionType> exceptionTypeRep,
  49. S8NotificationService notificationService,
  50. S8ManualReportService manualReportService,
  51. S8TimeoutRuleEvaluator timeoutEvaluator,
  52. S8ShortageRuleEvaluator shortageEvaluator,
  53. S8OutOfRangeRuleEvaluator outOfRangeEvaluator,
  54. ILogger<S8WatchSchedulerService> logger,
  55. SqlSugarRepository<AdoS8DetectionLog> detectionLogRep,
  56. SqlSugarRepository<AdoS8RuleDetectionState> detectionStateRep)
  57. {
  58. _ruleRep = ruleRep;
  59. _alertRuleRep = alertRuleRep;
  60. _dataSourceRep = dataSourceRep;
  61. _exceptionRep = exceptionRep;
  62. _exceptionTypeRep = exceptionTypeRep;
  63. _notificationService = notificationService;
  64. _manualReportService = manualReportService;
  65. _timeoutEvaluator = timeoutEvaluator;
  66. _shortageEvaluator = shortageEvaluator;
  67. _outOfRangeEvaluator = outOfRangeEvaluator;
  68. _logger = logger;
  69. _detectionLogRep = detectionLogRep;
  70. _detectionStateRep = detectionStateRep;
  71. }
  72. public async Task<List<S8WatchExecutionRule>> LoadExecutionRulesAsync(long tenantId, long factoryId)
  73. {
  74. // R3 OUT_OF_RANGE 重写后,新三类(OUT_OF_RANGE/TIMEOUT/SHORTAGE)改走 ProcessRulesByTypeAsync。
  75. // 旧 AlertRule 兼容主链此处只装载 RuleType 为空/未分类的历史规则,避免新旧双跑导致重复建单。
  76. var watchRules = await _ruleRep.AsQueryable()
  77. .Where(x => x.TenantId == tenantId
  78. && x.FactoryId == factoryId
  79. && x.Enabled
  80. && x.SceneCode == S8SceneCode.S2S6Production
  81. && (x.RuleType == null || x.RuleType == ""))
  82. .ToListAsync();
  83. var deviceRules = watchRules
  84. .Where(x => IsDeviceWatchObjectType(x.WatchObjectType))
  85. .ToList();
  86. if (deviceRules.Count == 0) return new();
  87. var dataSourceIds = deviceRules
  88. .Select(x => x.DataSourceId)
  89. .Distinct()
  90. .ToList();
  91. var dataSources = await _dataSourceRep.AsQueryable()
  92. .Where(x => x.TenantId == tenantId
  93. && x.FactoryId == factoryId
  94. && x.Enabled
  95. && dataSourceIds.Contains(x.Id))
  96. .ToListAsync();
  97. var dataSourceMap = dataSources.ToDictionary(x => x.Id);
  98. if (dataSourceMap.Count == 0) return new();
  99. var alertRules = (await _alertRuleRep.AsQueryable()
  100. .Where(x => x.TenantId == tenantId
  101. && x.FactoryId == factoryId
  102. && x.SceneCode == S8SceneCode.S2S6Production)
  103. .ToListAsync())
  104. .Where(IsSupportedAlertRule)
  105. .ToList();
  106. // G-01 首版 AlertRule 冲突口径(C 收口):
  107. // 当前场景存在多条可运行 AlertRule 时,视为“当前规则配置冲突”并跳过该规则,
  108. // 不按“首条”继续运行,也不扩大为“整场景停摆”。
  109. // 当前模型下所有 device watchRule 共享同场景 AlertRule,故冲突态下所有 device 规则均跳过,
  110. // 但此处按“逐规则跳过”的语义实现,避免被误读为“整场景 return empty 停摆”。
  111. var alertRule = alertRules.Count == 1 ? alertRules[0] : null;
  112. var executionRules = new List<S8WatchExecutionRule>();
  113. foreach (var watchRule in deviceRules.OrderBy(x => x.Id))
  114. {
  115. // 配置冲突:当前规则跳过(不停摆其他规则)。
  116. if (alertRule == null)
  117. continue;
  118. if (!dataSourceMap.TryGetValue(watchRule.DataSourceId, out var dataSource))
  119. continue;
  120. if (!IsSupportedSqlDataSource(dataSource))
  121. continue;
  122. executionRules.Add(new S8WatchExecutionRule
  123. {
  124. WatchRuleId = watchRule.Id,
  125. WatchRuleCode = watchRule.RuleCode,
  126. SceneCode = watchRule.SceneCode,
  127. TriggerType = DefaultTriggerType,
  128. WatchObjectType = watchRule.WatchObjectType.Trim(),
  129. DataSourceId = dataSource.Id,
  130. DataSourceCode = dataSource.DataSourceCode,
  131. DataSourceType = dataSource.Type,
  132. DataSourceConnection = dataSource.Endpoint?.Trim() ?? string.Empty,
  133. QueryExpression = watchRule.Expression?.Trim() ?? string.Empty,
  134. PollIntervalSeconds = watchRule.PollIntervalSeconds,
  135. AlertRuleId = alertRule.Id,
  136. AlertRuleCode = alertRule.RuleCode,
  137. TriggerCondition = alertRule.TriggerCondition!.Trim(),
  138. ThresholdValue = alertRule.ThresholdVal!.Trim(),
  139. Severity = alertRule.Severity
  140. });
  141. }
  142. return executionRules;
  143. }
  144. public async Task<List<S8WatchDeviceQueryResult>> QueryDeviceRowsAsync(long tenantId, long factoryId)
  145. {
  146. var executionRules = await LoadExecutionRulesAsync(tenantId, factoryId);
  147. var results = new List<S8WatchDeviceQueryResult>();
  148. foreach (var rule in executionRules)
  149. results.Add(await QueryDeviceRowsAsync(rule));
  150. return results;
  151. }
  152. public async Task<S8WatchDeviceQueryResult> QueryDeviceRowsAsync(S8WatchExecutionRule rule)
  153. {
  154. if (!string.Equals(rule.DataSourceType, SqlDataSourceType, StringComparison.OrdinalIgnoreCase))
  155. return S8WatchDeviceQueryResult.Fail(rule, "数据源类型不是 SQL,已跳过");
  156. if (string.IsNullOrWhiteSpace(rule.QueryExpression))
  157. return S8WatchDeviceQueryResult.Fail(rule, "查询表达式为空,已跳过");
  158. try
  159. {
  160. using var db = CreateSqlQueryScope(rule.DataSourceConnection);
  161. var table = await db.Ado.GetDataTableAsync(rule.QueryExpression);
  162. if (!HasRequiredColumns(table))
  163. return S8WatchDeviceQueryResult.Fail(rule, "查询结果缺少 required columns: related_object_code/current_value");
  164. var rows = table.Rows.Cast<DataRow>()
  165. .Select(MapDeviceRow)
  166. .Where(x => !string.IsNullOrWhiteSpace(x.RelatedObjectCode))
  167. .ToList();
  168. return S8WatchDeviceQueryResult.Ok(rule, rows);
  169. }
  170. catch (Exception ex)
  171. {
  172. return S8WatchDeviceQueryResult.Fail(rule, $"查询执行失败: {ex.Message}");
  173. }
  174. }
  175. /// <summary>
  176. /// G01-04:基于设备级结果行集做首版 VALUE_DEVIATION 单阈值判定,
  177. /// 产出命中结果对象列表,供 G01-05 去重与 G01-06 建单消费。
  178. /// 本方法不做去重、不做建单、不做严重度重算、不做时间线。
  179. /// </summary>
  180. public async Task<List<S8WatchHitResult>> EvaluateHitsAsync(long tenantId, long factoryId)
  181. {
  182. var executionRules = await LoadExecutionRulesAsync(tenantId, factoryId);
  183. var ruleMap = executionRules.ToDictionary(x => x.WatchRuleId);
  184. var queryResults = new List<S8WatchDeviceQueryResult>();
  185. foreach (var rule in executionRules)
  186. queryResults.Add(await QueryDeviceRowsAsync(rule));
  187. var hits = new List<S8WatchHitResult>();
  188. foreach (var queryResult in queryResults)
  189. {
  190. // G01-03 查询失败:跳过,不进入判定。
  191. if (!queryResult.Success) continue;
  192. if (!ruleMap.TryGetValue(queryResult.WatchRuleId, out var rule)) continue;
  193. // 判定参数缺失:跳过当前规则。
  194. if (string.IsNullOrWhiteSpace(rule.TriggerCondition)
  195. || string.IsNullOrWhiteSpace(rule.ThresholdValue))
  196. continue;
  197. // 比较符非法:跳过当前规则。
  198. var op = TryParseTriggerCondition(rule.TriggerCondition);
  199. if (op == null) continue;
  200. // ThresholdValue 非法:跳过当前规则。
  201. if (!TryParseDecimal(rule.ThresholdValue, out var threshold)) continue;
  202. foreach (var row in queryResult.Rows)
  203. {
  204. // CurrentValue 非法(null / 无法解析数值):跳过当前行,不进入判定。
  205. if (row.CurrentValue == null) continue;
  206. // 未命中:不进入后续链路。
  207. if (!EvaluateHit(row.CurrentValue.Value, op, threshold)) continue;
  208. hits.Add(new S8WatchHitResult
  209. {
  210. SourceRuleId = rule.WatchRuleId,
  211. SourceRuleCode = rule.WatchRuleCode,
  212. AlertRuleId = rule.AlertRuleId,
  213. DataSourceId = rule.DataSourceId,
  214. RelatedObjectCode = row.RelatedObjectCode,
  215. CurrentValue = row.CurrentValue.Value,
  216. ThresholdValue = threshold,
  217. TriggerCondition = op,
  218. Severity = rule.Severity,
  219. OccurrenceDeptId = row.OccurrenceDeptId,
  220. ResponsibleDeptId = row.ResponsibleDeptId,
  221. SourcePayload = row.SourcePayload
  222. });
  223. }
  224. }
  225. return hits;
  226. }
  227. /// <summary>
  228. /// G01-05:未闭环异常去重最小实现。
  229. /// 消费 G01-04 产出的 <see cref="S8WatchHitResult"/>,按 (SourceRuleId + RelatedObjectCode)
  230. /// 在未闭环状态集合内判重,只回答“是否允许建单”。
  231. /// 首版明确不做:原单刷新 / 时间线追加 / payload 更新 / 次数累计 / 严重度重算 / 状态修复。
  232. /// </summary>
  233. public async Task<List<S8WatchDedupResult>> EvaluateDedupAsync(long tenantId, long factoryId)
  234. {
  235. var hits = await EvaluateHitsAsync(tenantId, factoryId);
  236. var results = new List<S8WatchDedupResult>(hits.Count);
  237. foreach (var hit in hits)
  238. {
  239. // 防御性分支:正常情况下 SourceRuleId 与 RelatedObjectCode 已由上游
  240. // (G01-02 规则装配 + G01-03 查询结果列校验)保证;此处仅作兜底,
  241. // 不是首版正常路径。
  242. if (hit.SourceRuleId <= 0 || string.IsNullOrWhiteSpace(hit.RelatedObjectCode))
  243. {
  244. results.Add(new S8WatchDedupResult
  245. {
  246. Hit = hit,
  247. CanCreate = false,
  248. MatchedExceptionId = null,
  249. Reason = "missing_dedup_key"
  250. });
  251. continue;
  252. }
  253. long matchedId;
  254. try
  255. {
  256. matchedId = await FindPendingExceptionIdAsync(
  257. tenantId, factoryId, hit.SourceRuleId, hit.RelatedObjectCode);
  258. }
  259. catch
  260. {
  261. // 去重查询失败:首版偏保守,宁可阻止也不重复建单,不扩展为补偿。
  262. results.Add(new S8WatchDedupResult
  263. {
  264. Hit = hit,
  265. CanCreate = false,
  266. MatchedExceptionId = null,
  267. Reason = "query_failed"
  268. });
  269. continue;
  270. }
  271. if (matchedId > 0)
  272. {
  273. // 命中已有未闭环异常 → 阻止建单。
  274. results.Add(new S8WatchDedupResult
  275. {
  276. Hit = hit,
  277. CanCreate = false,
  278. MatchedExceptionId = matchedId,
  279. Reason = "duplicate_pending"
  280. });
  281. }
  282. else
  283. {
  284. // 未命中 → 允许建单,交 G01-06 消费。
  285. results.Add(new S8WatchDedupResult
  286. {
  287. Hit = hit,
  288. CanCreate = true,
  289. MatchedExceptionId = null,
  290. Reason = "no_pending"
  291. });
  292. }
  293. }
  294. return results;
  295. }
  296. // 取任意一条匹配的未闭环异常 Id 作为“是否存在重复单”的拦截依据。
  297. // 首版只需要“存在性”,不关心“最早 / 最新”;不在 G01-05 处理排序语义。
  298. private async Task<long> FindPendingExceptionIdAsync(
  299. long tenantId, long factoryId, long sourceRuleId, string relatedObjectCode)
  300. {
  301. // SqlSugar 表达式翻译要求 Contains 数组变量必须可访问;此处将类级常量
  302. // 承接到方法内局部变量,仅为表达式翻译服务,值与 UnclosedExceptionStatuses 一致。
  303. var statuses = UnclosedExceptionStatuses;
  304. var ids = await _exceptionRep.AsQueryable()
  305. .Where(x => x.TenantId == tenantId
  306. && x.FactoryId == factoryId
  307. && !x.IsDeleted
  308. && x.SourceRuleId == sourceRuleId
  309. && x.RelatedObjectCode == relatedObjectCode
  310. && statuses.Contains(x.Status))
  311. .Select(x => x.Id)
  312. .Take(1)
  313. .ToListAsync();
  314. return ids.Count > 0 ? ids[0] : 0L;
  315. }
  316. /// <summary>
  317. /// G01-06:自动建单入口。消费 G01-05 去重结果,对 CanCreate==true 的命中
  318. /// 复用 S8ManualReportService.CreateFromWatchAsync(同一主链的自动建单分支)落成标准 AdoS8Exception。
  319. /// CanCreate==false 直接跳过;创建失败返回最小失败结果,不补偿、不重试、不对账。
  320. /// </summary>
  321. public async Task<List<S8WatchCreationResult>> CreateExceptionsAsync(long tenantId, long factoryId)
  322. {
  323. var dedupResults = await EvaluateDedupAsync(tenantId, factoryId);
  324. var results = new List<S8WatchCreationResult>(dedupResults.Count);
  325. foreach (var dedup in dedupResults)
  326. {
  327. if (!dedup.CanCreate)
  328. {
  329. results.Add(new S8WatchCreationResult
  330. {
  331. DedupResult = dedup,
  332. Created = false,
  333. Skipped = true,
  334. CreatedExceptionId = null,
  335. Reason = dedup.Reason,
  336. ErrorMessage = null
  337. });
  338. continue;
  339. }
  340. try
  341. {
  342. var entity = await _manualReportService.CreateFromWatchAsync(dedup.Hit);
  343. results.Add(new S8WatchCreationResult
  344. {
  345. DedupResult = dedup,
  346. Created = true,
  347. Skipped = false,
  348. CreatedExceptionId = entity.Id,
  349. Reason = "auto_created",
  350. ErrorMessage = null
  351. });
  352. }
  353. catch (Exception ex)
  354. {
  355. results.Add(new S8WatchCreationResult
  356. {
  357. DedupResult = dedup,
  358. Created = false,
  359. Skipped = false,
  360. CreatedExceptionId = null,
  361. Reason = "create_failed",
  362. ErrorMessage = ex.Message
  363. });
  364. }
  365. }
  366. // R3-OUT_OF_RANGE-REWRITE-1:三类正式 evaluator 路径。
  367. // 旧 AlertRule 兼容主链(上方 dedupResults 循环)已在 LoadExecutionRulesAsync 处过滤掉
  368. // OUT_OF_RANGE/TIMEOUT/SHORTAGE,只保留未分类历史规则;故新旧不双跑。
  369. // R6 RunId:本次 CreateExceptionsAsync 调用对应的统一关联 id,落入 detection_log。
  370. var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
  371. results.AddRange(await ProcessRulesByTypeAsync(tenantId, factoryId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, runId));
  372. results.AddRange(await ProcessRulesByTypeAsync(tenantId, factoryId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, runId));
  373. results.AddRange(await ProcessRulesByTypeAsync(tenantId, factoryId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, runId));
  374. return results;
  375. }
  376. /// <summary>
  377. /// R2 TIMEOUT 类规则主链:薄包装,复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
  378. /// </summary>
  379. public Task<List<S8WatchCreationResult>> ProcessTimeoutRulesAsync(long tenantId, long factoryId) =>
  380. ProcessRulesByTypeAsync(tenantId, factoryId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  381. /// <summary>
  382. /// R3 SHORTAGE 类规则主链:薄包装,复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
  383. /// </summary>
  384. public Task<List<S8WatchCreationResult>> ProcessShortageRulesAsync(long tenantId, long factoryId) =>
  385. ProcessRulesByTypeAsync(tenantId, factoryId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  386. /// <summary>
  387. /// R3-OUT_OF_RANGE-REWRITE-1:OUT_OF_RANGE 类规则主链。
  388. /// 复用 <see cref="ProcessRulesByTypeAsync"/>,并对历史 dedup_key=NULL 的旧记录做 compat fallback:
  389. /// (source_rule_id=rule.Id AND related_object_code=hit AND status!=CLOSED AND dedup_key IS NULL AND is_deleted=0)
  390. /// 命中则 backfill 6 列,避免重复建单。RunId 由内部生成。
  391. /// </summary>
  392. public Task<List<S8WatchCreationResult>> ProcessOutOfRangeRulesAsync(long tenantId, long factoryId) =>
  393. ProcessRulesByTypeAsync(tenantId, factoryId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));
  394. /// <summary>
  395. /// R2/R3 通用规则主链。S8-SCHED-CLEANUP-LEGACY-PATH-1:本方法已收敛为 thin wrapper,
  396. /// 单规则处理(evaluator → reconcile → hit 循环 → CREATED/REFRESHED/NO_HIT/EVALUATE_FAILED 日志)
  397. /// 全部下沉至 <see cref="ProcessSingleRuleAsync"/>,与 Job tick 路径
  398. /// (<see cref="RunDispatchTickAsync"/> → <see cref="RunSingleRuleAsync"/> → ProcessSingleRuleAsync)
  399. /// 共享同一份逻辑,避免双维护。
  400. ///
  401. /// 调用方仅有 <see cref="CreateExceptionsAsync"/>(debug run-once / Process*RulesAsync 公共薄包装)。
  402. /// debug run-once 不持 lease(保留"手动立即跑"语义),但本方法在每条规则结束时通过
  403. /// <see cref="ApplyRunOnceCompletionAsync"/> 同步更新 watch_rule.last_run_at / last_status /
  404. /// last_error / last_duration_ms / last_run_id / consecutive_failure_count,缩小 run-once 与
  405. /// Job tick 之间的 last_* 状态分裂;不动 lock_token / running_started_at / paused_until,避免
  406. /// 与正在持锁运行的 Job 撕扯。
  407. /// </summary>
  408. private async Task<List<S8WatchCreationResult>> ProcessRulesByTypeAsync(
  409. long tenantId, long factoryId, IS8RuleEvaluator evaluator, string ruleType, string runId)
  410. {
  411. var aggregate = new List<S8WatchCreationResult>();
  412. var rules = await _ruleRep.AsQueryable()
  413. .Where(x => x.TenantId == tenantId
  414. && x.FactoryId == factoryId
  415. && x.Enabled
  416. && x.RuleType == ruleType)
  417. .ToListAsync();
  418. if (rules.Count == 0) return aggregate;
  419. var alertRules = (await _alertRuleRep.AsQueryable()
  420. .Where(x => x.TenantId == tenantId && x.FactoryId == factoryId)
  421. .ToListAsync()).AsReadOnly();
  422. foreach (var rule in rules.OrderBy(x => x.Id))
  423. {
  424. var sw = System.Diagnostics.Stopwatch.StartNew();
  425. S8RuleRunResult completion;
  426. try
  427. {
  428. var ruleResults = await ProcessSingleRuleAsync(tenantId, factoryId, rule, ruleType, evaluator, alertRules, runId);
  429. aggregate.AddRange(ruleResults);
  430. completion = new S8RuleRunResult
  431. {
  432. Success = true,
  433. Stats = new S8RuleRunStats
  434. {
  435. Hits = ruleResults.Count,
  436. Created = ruleResults.Count(r => r.Created),
  437. Refreshed = ruleResults.Count(r => r.Reason == "duplicate_pending"),
  438. Pending = ruleResults.Count(r => r.Reason == "antiflap_pending_hit"),
  439. Failed = ruleResults.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
  440. }
  441. };
  442. }
  443. catch (Exception ex)
  444. {
  445. // ProcessSingleRuleAsync 在 evaluator 抛 S8RuleEvaluatorException 时已写 EVALUATE_FAILED
  446. // detection_log 后再 throw(供 RunSingleRuleAsync 标 Success=false)。本路径无 lease,
  447. // 吞异常以保留"逐规则失败不影响其他规则"的旧 ProcessRulesByTypeAsync 语义。
  448. _logger.LogWarning(ex, "process_rule_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  449. aggregate.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
  450. completion = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
  451. }
  452. sw.Stop();
  453. try
  454. {
  455. await ApplyRunOnceCompletionAsync(rule, completion, (int)sw.ElapsedMilliseconds, runId);
  456. }
  457. catch (Exception ex)
  458. {
  459. _logger.LogWarning(ex, "run_once_completion_write_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  460. }
  461. }
  462. return aggregate;
  463. }
  464. /// <summary>
  465. /// S8-SCHED-CLEANUP-LEGACY-PATH-1:debug run-once 的 last_* 状态回写(无 lease 版)。
  466. /// 与 <see cref="OnRuleCompletedAsync"/> 的语义平行,差异在于:
  467. /// - 不要求 lock_token 匹配(run-once 不持 lease)
  468. /// - 不写 lock_token / locked_by / lock_until / running_started_at(不与 Job lease 撕扯)
  469. /// - 不做 consecutive_failure_count 阈值的 paused_until 自动暂停(debug 路径不应自动暂停 demo rule)
  470. /// 写入:last_run_at / next_run_at / last_status / last_error / last_duration_ms / last_run_id /
  471. /// consecutive_failure_count / updated_at。
  472. /// </summary>
  473. private async Task ApplyRunOnceCompletionAsync(AdoS8WatchRule rule, S8RuleRunResult result, int durationMs, string runId)
  474. {
  475. var now = DateTime.Now;
  476. var nextRunAt = now.AddSeconds(NormalizePollInterval(rule.PollIntervalSeconds));
  477. if (result.Success)
  478. {
  479. await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  480. .SetColumns(x => new AdoS8WatchRule
  481. {
  482. LastRunAt = now,
  483. NextRunAt = nextRunAt,
  484. LastStatus = "SUCCESS",
  485. LastError = null,
  486. LastDurationMs = durationMs,
  487. LastRunId = runId,
  488. ConsecutiveFailureCount = 0,
  489. UpdatedAt = now
  490. })
  491. .Where(x => x.Id == rule.Id)
  492. .ExecuteCommandAsync();
  493. }
  494. else
  495. {
  496. var errorTrunc = Truncate(result.ErrorMessage, 500);
  497. await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  498. .SetColumns(x => new AdoS8WatchRule
  499. {
  500. LastRunAt = now,
  501. NextRunAt = nextRunAt,
  502. LastStatus = "FAILED",
  503. LastError = errorTrunc,
  504. LastDurationMs = durationMs,
  505. LastRunId = runId,
  506. ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
  507. UpdatedAt = now
  508. })
  509. .Where(x => x.Id == rule.Id)
  510. .ExecuteCommandAsync();
  511. }
  512. }
  513. // S8-SCHED-EXEC-1:trigger / recover 抗抖计数兜底,null / <1 / >10 一律按 1,避免非法配置导致永远不建单 / 永远不恢复。
  514. private static int NormalizeAntiflapCount(int raw)
  515. {
  516. if (raw < 1 || raw > 10) return 1;
  517. return raw;
  518. }
  519. /// <summary>
  520. /// 命中时累加 detection_state.consecutive_hit_count;未存在则插入 hitCount=1。
  521. /// 返回当前 state 行(含 Id)以及命中后的 hitCount。
  522. /// 注意:本函数不消费 trigger_count_required;上游决定是否进入 CreateFromHitAsync。
  523. /// </summary>
  524. private async Task<(AdoS8RuleDetectionState? state, int hitCount)> UpsertDetectionStateOnHitAsync(
  525. long tenantId, long factoryId, AdoS8WatchRule rule, S8RuleHit hit)
  526. {
  527. var now = DateTime.Now;
  528. var existing = await _detectionStateRep.AsQueryable()
  529. .Where(x => x.TenantId == tenantId
  530. && x.FactoryId == factoryId
  531. && x.RuleCode == rule.RuleCode
  532. && x.DedupKey == hit.DedupKey)
  533. .FirstAsync();
  534. if (existing == null)
  535. {
  536. var fresh = new AdoS8RuleDetectionState
  537. {
  538. TenantId = tenantId,
  539. FactoryId = factoryId,
  540. RuleCode = rule.RuleCode,
  541. DedupKey = hit.DedupKey,
  542. SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? null : hit.SourceObjectType,
  543. SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? null : hit.SourceObjectId,
  544. ConsecutiveHitCount = 1,
  545. ConsecutiveMissCount = 0,
  546. LastSeenAt = now,
  547. LastHitAt = now,
  548. CreatedAt = now,
  549. UpdatedAt = now
  550. };
  551. // BUG-S8-DETECTION-STATE-ACTIVE-EXC-ID-TRIGGER1-001:必须回填 fresh.Id,否则
  552. // trigger=1 首 tick 建单后 UPDATE state SET active_exception_id WHERE id=state.Id
  553. // 命中 0 行(state.Id 为默认 0)。沿用 S8ManualReportService 既定模式。
  554. fresh = await _detectionStateRep.AsInsertable(fresh).ExecuteReturnEntityAsync();
  555. return (fresh, 1);
  556. }
  557. var newHitCount = existing.ConsecutiveHitCount + 1;
  558. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  559. .SetColumns(x => new AdoS8RuleDetectionState
  560. {
  561. ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
  562. ConsecutiveMissCount = 0,
  563. LastSeenAt = now,
  564. LastHitAt = now,
  565. SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? existing.SourceObjectType : hit.SourceObjectType,
  566. SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? existing.SourceObjectId : hit.SourceObjectId,
  567. UpdatedAt = now
  568. })
  569. .Where(x => x.Id == existing.Id)
  570. .ExecuteCommandAsync();
  571. existing.ConsecutiveHitCount = newHitCount;
  572. return (existing, newHitCount);
  573. }
  574. private async Task<long> FindOpenExceptionByDedupKeyAsync(long tenantId, long factoryId, string dedupKey)
  575. {
  576. var ids = await _exceptionRep.AsQueryable()
  577. .Where(x => x.TenantId == tenantId
  578. && x.FactoryId == factoryId
  579. && !x.IsDeleted
  580. && x.Status != "CLOSED"
  581. && x.DedupKey == dedupKey)
  582. .Select(x => x.Id)
  583. .Take(1)
  584. .ToListAsync();
  585. return ids.Count > 0 ? ids[0] : 0L;
  586. }
  587. /// <summary>
  588. /// R3 OUT_OF_RANGE compat fallback 查找:用 (source_rule_id, related_object_code, status!=CLOSED,
  589. /// dedup_key IS NULL, is_deleted=0) 严格条件定位旧 AlertRule 主链留下的历史记录。
  590. /// </summary>
  591. private async Task<long> FindLegacyOutOfRangeExceptionAsync(long tenantId, long factoryId, long sourceRuleId, string relatedObjectCode)
  592. {
  593. if (string.IsNullOrWhiteSpace(relatedObjectCode)) return 0L;
  594. var ids = await _exceptionRep.AsQueryable()
  595. .Where(x => x.TenantId == tenantId
  596. && x.FactoryId == factoryId
  597. && !x.IsDeleted
  598. && x.Status != "CLOSED"
  599. && x.DedupKey == null
  600. && x.SourceRuleId == sourceRuleId
  601. && x.RelatedObjectCode == relatedObjectCode)
  602. .Select(x => x.Id)
  603. .Take(1)
  604. .ToListAsync();
  605. return ids.Count > 0 ? ids[0] : 0L;
  606. }
  607. /// <summary>
  608. /// R3 OUT_OF_RANGE compat fallback backfill:把历史记录的 R1 新 6 列(dedup_key/source_rule_code/
  609. /// source_object_type/source_object_id/source_payload/last_detected_at)写入,并刷新 updated_at。
  610. /// </summary>
  611. private async Task BackfillLegacyExceptionAsync(long exceptionId, S8RuleHit hit)
  612. {
  613. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  614. .SetColumns(x => new AdoS8Exception
  615. {
  616. DedupKey = hit.DedupKey,
  617. SourceRuleCode = hit.SourceRuleCode,
  618. SourceObjectType = hit.SourceObjectType,
  619. SourceObjectId = hit.SourceObjectId,
  620. SourcePayload = hit.SourcePayload,
  621. LastDetectedAt = hit.DetectedAt,
  622. UpdatedAt = DateTime.Now
  623. })
  624. .Where(x => x.Id == exceptionId)
  625. .ExecuteCommandAsync();
  626. }
  627. /// <summary>
  628. /// R5 恢复时间最小闭环:对当前 rule 下未关闭、有 dedup_key、recovered_at 仍为 NULL 的异常,
  629. /// 凡不在本轮 hits.dedup_key 集合内的,写入 recovered_at = now、updated_at = now。
  630. /// 仅写这 2 列;不动 status / assignee / verifier / source_payload / last_detected_at;
  631. /// recovered_at 一旦写入,本轮不做复发清空。
  632. /// R6 返回 recoveredIds 供上游决定是否写 NO_HIT 日志,并对每个 recovered exception 写一条 RECOVERED 日志。
  633. /// </summary>
  634. private async Task<List<long>> ReconcileRecoveriesForRuleAsync(
  635. long tenantId, long factoryId, AdoS8WatchRule rule, string ruleType, List<S8RuleHit> hits, string runId)
  636. {
  637. var hitDedupKeys = hits
  638. .Where(h => !string.IsNullOrWhiteSpace(h.DedupKey))
  639. .Select(h => h.DedupKey)
  640. .ToHashSet(StringComparer.Ordinal);
  641. var candidates = await _exceptionRep.AsQueryable()
  642. .Where(x => x.TenantId == tenantId
  643. && x.FactoryId == factoryId
  644. && !x.IsDeleted
  645. && x.Status != "CLOSED"
  646. && x.SourceRuleCode == rule.RuleCode
  647. && x.DedupKey != null
  648. && x.RecoveredAt == null)
  649. .Select(x => new { x.Id, x.DedupKey, x.SourceObjectType, x.SourceObjectId, x.RelatedObjectCode, x.ConsecutiveMissCount })
  650. .ToListAsync();
  651. if (candidates.Count == 0) return new List<long>();
  652. var now = DateTime.Now;
  653. var recoverRequired = NormalizeAntiflapCount(rule.RecoverCountRequired);
  654. var recoveredIds = new List<long>();
  655. foreach (var c in candidates)
  656. {
  657. if (hitDedupKeys.Contains(c.DedupKey!)) continue;
  658. // S8-SCHED-EXEC-1:恢复抗抖累计。
  659. // 1) 每次未命中:异常 ConsecutiveMissCount += 1,ConsecutiveHitCount 清零;
  660. // 2) miss < recover_count_required:仅累计,不写 recovered_at、不写 RECOVERED;
  661. // 3) miss >= recover_count_required:写 recovered_at、写 RECOVERED 日志。
  662. var newMissCount = c.ConsecutiveMissCount + 1;
  663. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  664. .SetColumns(x => new AdoS8Exception
  665. {
  666. ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
  667. ConsecutiveHitCount = 0,
  668. UpdatedAt = now
  669. })
  670. .Where(x => x.Id == c.Id)
  671. .ExecuteCommandAsync();
  672. // detection_state 同步累计 miss(建单后 state.active_exception_id 仍指向 c.Id)。
  673. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  674. .SetColumns(x => new AdoS8RuleDetectionState
  675. {
  676. ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
  677. ConsecutiveHitCount = 0,
  678. LastSeenAt = now,
  679. LastMissAt = now,
  680. UpdatedAt = now
  681. })
  682. .Where(x => x.TenantId == tenantId
  683. && x.FactoryId == factoryId
  684. && x.RuleCode == rule.RuleCode
  685. && x.DedupKey == c.DedupKey)
  686. .ExecuteCommandAsync();
  687. if (newMissCount < recoverRequired)
  688. {
  689. _logger.LogInformation(
  690. "antiflap_pending_recovery ruleCode={RuleCode} dedupKey={DedupKey} missCount={Miss} recoverRequired={Required}",
  691. rule.RuleCode, c.DedupKey, newMissCount, recoverRequired);
  692. continue;
  693. }
  694. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  695. .SetColumns(x => new AdoS8Exception
  696. {
  697. RecoveredAt = now,
  698. UpdatedAt = now
  699. })
  700. .Where(x => x.Id == c.Id)
  701. .ExecuteCommandAsync();
  702. recoveredIds.Add(c.Id);
  703. await WriteDetectionLogAsync(new AdoS8DetectionLog
  704. {
  705. TenantId = tenantId, FactoryId = factoryId,
  706. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  707. SourceObjectType = c.SourceObjectType, SourceObjectId = c.SourceObjectId,
  708. RelatedObjectCode = c.RelatedObjectCode, DedupKey = c.DedupKey,
  709. DetectResult = DetectResultRecovered,
  710. ExceptionId = c.Id,
  711. DetectedAt = now,
  712. PayloadSnapshot = JsonSerializer.Serialize(new { ruleId = rule.Id, ruleCode = rule.RuleCode, reason = "no_longer_hit", missCount = newMissCount, recoverRequired }),
  713. RunId = runId, TriggerSource = DetectionTriggerSource,
  714. Remark = "Rule no longer hit; recovered_at marked"
  715. });
  716. }
  717. if (recoveredIds.Count > 0)
  718. {
  719. _logger.LogInformation(
  720. "rule_recovered ruleCode={RuleCode} ruleType={RuleType} recoveredCount={Count} recoveredIds={Ids}",
  721. rule.RuleCode, ruleType, recoveredIds.Count, string.Join(",", recoveredIds));
  722. }
  723. return recoveredIds;
  724. }
  725. /// <summary>R6 通用 hit 日志构造(CREATED / REFRESHED 共用)。</summary>
  726. private static AdoS8DetectionLog BuildHitLog(
  727. long tenantId, long factoryId, AdoS8WatchRule rule, string ruleType, S8RuleHit hit,
  728. string detectResult, long exceptionId, string runId) => new()
  729. {
  730. TenantId = tenantId, FactoryId = factoryId,
  731. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  732. SourceObjectType = hit.SourceObjectType, SourceObjectId = hit.SourceObjectId,
  733. RelatedObjectCode = hit.RelatedObjectCode, DedupKey = hit.DedupKey,
  734. DetectResult = detectResult,
  735. ExceptionId = exceptionId,
  736. DetectedAt = hit.DetectedAt,
  737. PayloadSnapshot = hit.SourcePayload,
  738. RunId = runId,
  739. TriggerSource = DetectionTriggerSource
  740. };
  741. /// <summary>R6 日志写入:失败仅 LogWarning,不阻断主链;不抛异常。</summary>
  742. private async Task WriteDetectionLogAsync(AdoS8DetectionLog log)
  743. {
  744. try
  745. {
  746. await _detectionLogRep.InsertAsync(log);
  747. }
  748. catch (Exception ex)
  749. {
  750. _logger.LogWarning(ex,
  751. "detection_log_write_failed runId={RunId} detectResult={Result} ruleCode={RuleCode} exceptionId={ExceptionId}",
  752. log.RunId, log.DetectResult, log.RuleCode, log.ExceptionId);
  753. }
  754. }
  755. private static string Truncate(string? s, int max) =>
  756. string.IsNullOrEmpty(s) ? string.Empty : (s.Length <= max ? s : s.Substring(0, max));
  757. private async Task RefreshDetectionAsync(long exceptionId, S8RuleHit hit)
  758. {
  759. // S8-SCHED-EXEC-1:刷新阶段抗抖累计 + 复发清空 recovered_at。
  760. // ConsecutiveHitCount += 1(用 SetColumns 内表达式完成原子自增);ConsecutiveMissCount 归零。
  761. // RecoveredAt 不为 NULL 时(复发)一并清空,保持业务对"再次命中即视为活跃"的预期。
  762. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  763. .SetColumns(x => new AdoS8Exception
  764. {
  765. LastDetectedAt = hit.DetectedAt,
  766. SourcePayload = hit.SourcePayload,
  767. ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
  768. ConsecutiveMissCount = 0,
  769. RecoveredAt = null,
  770. UpdatedAt = DateTime.Now
  771. })
  772. .Where(x => x.Id == exceptionId)
  773. .ExecuteCommandAsync();
  774. }
  775. private static S8WatchCreationResult BuildCreatedResult(AdoS8WatchRule rule, S8RuleHit hit, long exceptionId) =>
  776. new()
  777. {
  778. DedupResult = new S8WatchDedupResult
  779. {
  780. Hit = ToWatchHit(rule, hit),
  781. CanCreate = true,
  782. MatchedExceptionId = null,
  783. Reason = "no_pending"
  784. },
  785. Created = true,
  786. Skipped = false,
  787. CreatedExceptionId = exceptionId,
  788. Reason = "auto_created",
  789. ErrorMessage = null
  790. };
  791. private static S8WatchCreationResult BuildSkippedDuplicate(AdoS8WatchRule rule, S8RuleHit hit, long matchedId) =>
  792. new()
  793. {
  794. DedupResult = new S8WatchDedupResult
  795. {
  796. Hit = ToWatchHit(rule, hit),
  797. CanCreate = false,
  798. MatchedExceptionId = matchedId,
  799. Reason = "duplicate_pending"
  800. },
  801. Created = false,
  802. Skipped = true,
  803. CreatedExceptionId = null,
  804. Reason = "duplicate_pending",
  805. ErrorMessage = null
  806. };
  807. private static S8WatchCreationResult BuildSkipResult(AdoS8WatchRule rule, string reason, string? error, S8RuleHit? hit = null) =>
  808. new()
  809. {
  810. DedupResult = new S8WatchDedupResult
  811. {
  812. Hit = hit != null ? ToWatchHit(rule, hit) : new S8WatchHitResult { SourceRuleId = rule.Id, SourceRuleCode = rule.RuleCode },
  813. CanCreate = false,
  814. MatchedExceptionId = null,
  815. Reason = reason
  816. },
  817. Created = false,
  818. Skipped = true,
  819. CreatedExceptionId = null,
  820. Reason = reason,
  821. ErrorMessage = error
  822. };
  823. private static S8WatchHitResult ToWatchHit(AdoS8WatchRule rule, S8RuleHit hit) => new()
  824. {
  825. SourceRuleId = hit.SourceRuleId == 0 ? rule.Id : hit.SourceRuleId,
  826. SourceRuleCode = string.IsNullOrEmpty(hit.SourceRuleCode) ? rule.RuleCode : hit.SourceRuleCode,
  827. DataSourceId = hit.DataSourceId,
  828. RelatedObjectCode = hit.RelatedObjectCode,
  829. Severity = hit.Severity,
  830. OccurrenceDeptId = hit.OccurrenceDeptId,
  831. ResponsibleDeptId = hit.ResponsibleDeptId,
  832. SourcePayload = hit.SourcePayload
  833. };
  834. /// <summary>
  835. /// 单次轮询入口。当前仅完成规则读取与组装,返回可执行规则数量,不做实际数据采集。
  836. /// </summary>
  837. public async Task<int> RunOnceAsync()
  838. {
  839. var executionRules = await LoadExecutionRulesAsync(1, 1);
  840. return executionRules.Count;
  841. }
  842. // G01-04 首版最小比较符集合:>, >=, <, <=。
  843. // 允许首尾空格;非此集合的一律视为“比较符非法”,由调用方跳过。
  844. private static string? TryParseTriggerCondition(string raw)
  845. {
  846. var normalized = raw.Trim();
  847. return normalized switch
  848. {
  849. ">" => ">",
  850. ">=" => ">=",
  851. "<" => "<",
  852. "<=" => "<=",
  853. _ => null
  854. };
  855. }
  856. private static bool TryParseDecimal(string raw, out decimal value) =>
  857. decimal.TryParse(raw.Trim(), NumberStyles.Any, CultureInfo.InvariantCulture, out value);
  858. private static bool EvaluateHit(decimal current, string op, decimal threshold) => op switch
  859. {
  860. ">" => current > threshold,
  861. ">=" => current >= threshold,
  862. "<" => current < threshold,
  863. "<=" => current <= threshold,
  864. _ => false
  865. };
  866. private static bool IsSupportedAlertRule(AdoS8AlertRule alertRule) =>
  867. !string.IsNullOrWhiteSpace(alertRule.TriggerCondition)
  868. && !string.IsNullOrWhiteSpace(alertRule.ThresholdVal);
  869. private static bool IsSupportedSqlDataSource(AdoS8DataSource dataSource) =>
  870. dataSource.Enabled
  871. && string.Equals(dataSource.Type?.Trim(), SqlDataSourceType, StringComparison.OrdinalIgnoreCase)
  872. && !string.IsNullOrWhiteSpace(dataSource.Endpoint);
  873. private static bool IsDeviceWatchObjectType(string? watchObjectType)
  874. {
  875. if (string.IsNullOrWhiteSpace(watchObjectType)) return false;
  876. var normalized = watchObjectType.Trim().ToUpperInvariant();
  877. return normalized is "DEVICE" or "EQUIPMENT" || watchObjectType.Trim() == "设备";
  878. }
  879. private SqlSugarScope CreateSqlQueryScope(string connectionString)
  880. {
  881. var dbType = _ruleRep.Context.CurrentConnectionConfig.DbType;
  882. return new SqlSugarScope(new ConnectionConfig
  883. {
  884. ConfigId = $"s8-watch-sql-{Guid.NewGuid():N}",
  885. DbType = dbType,
  886. ConnectionString = connectionString,
  887. InitKeyType = InitKeyType.Attribute,
  888. IsAutoCloseConnection = true
  889. });
  890. }
  891. private static bool HasRequiredColumns(DataTable table) =>
  892. TryGetColumnName(table.Columns, "related_object_code") != null
  893. && TryGetColumnName(table.Columns, "current_value") != null;
  894. private static S8WatchDeviceRow MapDeviceRow(DataRow row)
  895. {
  896. var columns = row.Table.Columns;
  897. var relatedObjectCodeColumn = TryGetColumnName(columns, "related_object_code");
  898. var currentValueColumn = TryGetColumnName(columns, "current_value");
  899. var occurrenceDeptIdColumn = TryGetColumnName(columns, "occurrence_dept_id");
  900. var responsibleDeptIdColumn = TryGetColumnName(columns, "responsible_dept_id");
  901. return new S8WatchDeviceRow
  902. {
  903. RelatedObjectCode = ReadString(row, relatedObjectCodeColumn),
  904. CurrentValue = ReadDecimal(row, currentValueColumn),
  905. OccurrenceDeptId = ReadLong(row, occurrenceDeptIdColumn),
  906. ResponsibleDeptId = ReadLong(row, responsibleDeptIdColumn),
  907. SourcePayload = BuildSourcePayload(row)
  908. };
  909. }
  910. private static string? TryGetColumnName(DataColumnCollection columns, string expectedName)
  911. {
  912. var normalizedExpected = NormalizeColumnName(expectedName);
  913. foreach (DataColumn column in columns)
  914. {
  915. if (NormalizeColumnName(column.ColumnName) == normalizedExpected)
  916. return column.ColumnName;
  917. }
  918. return null;
  919. }
  920. private static string NormalizeColumnName(string columnName) =>
  921. columnName.Replace("_", string.Empty, StringComparison.Ordinal).Trim().ToUpperInvariant();
  922. private static string ReadString(DataRow row, string? columnName)
  923. {
  924. if (string.IsNullOrWhiteSpace(columnName)) return string.Empty;
  925. var value = row[columnName];
  926. return value == DBNull.Value ? string.Empty : Convert.ToString(value)?.Trim() ?? string.Empty;
  927. }
  928. private static long? ReadLong(DataRow row, string? columnName)
  929. {
  930. if (string.IsNullOrWhiteSpace(columnName)) return null;
  931. var value = row[columnName];
  932. if (value == DBNull.Value) return null;
  933. return long.TryParse(Convert.ToString(value, CultureInfo.InvariantCulture), out var result) ? result : null;
  934. }
  935. private static decimal? ReadDecimal(DataRow row, string? columnName)
  936. {
  937. if (string.IsNullOrWhiteSpace(columnName)) return null;
  938. var value = row[columnName];
  939. if (value == DBNull.Value) return null;
  940. return decimal.TryParse(Convert.ToString(value, CultureInfo.InvariantCulture), NumberStyles.Any, CultureInfo.InvariantCulture, out var result)
  941. ? result
  942. : null;
  943. }
  944. private static string BuildSourcePayload(DataRow row)
  945. {
  946. var payload = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
  947. foreach (DataColumn column in row.Table.Columns)
  948. {
  949. var value = row[column];
  950. payload[column.ColumnName] = value == DBNull.Value ? null : value;
  951. }
  952. return JsonSerializer.Serialize(payload);
  953. }
  954. // ============================================================
  955. // S8-SCHED-EXEC-1:DB 驱动调度执行层
  956. // ============================================================
  957. private const int LeaseDurationMinutes = 5;
  958. private const int AutoPauseFailureThreshold = 3;
  959. private const int AutoPauseDurationHours = 1;
  960. private const int DefaultPollIntervalSeconds = 300;
  961. private const int MinPollIntervalSeconds = 60;
  962. private const int MaxPollIntervalSeconds = 86400;
  963. /// <summary>
  964. /// S8-SCHED-EXEC-1:释放过期 lease(lock_until &lt; NOW),不修改 last_status / last_error,仅清空 lock 三件套 + running_started_at。
  965. /// 返回释放的行数。
  966. /// </summary>
  967. public async Task<int> ResetExpiredLeasesAsync(long tenantId, long factoryId)
  968. {
  969. var now = DateTime.Now;
  970. var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  971. .SetColumns(x => new AdoS8WatchRule
  972. {
  973. LockToken = null,
  974. LockedBy = null,
  975. LockUntil = null,
  976. RunningStartedAt = null,
  977. UpdatedAt = now
  978. })
  979. .Where(x => x.TenantId == tenantId
  980. && x.FactoryId == factoryId
  981. && x.LockUntil != null
  982. && x.LockUntil < now)
  983. .ExecuteCommandAsync();
  984. if (affected > 0)
  985. {
  986. _logger.LogWarning(
  987. "lease_reset tenantId={Tenant} factoryId={Factory} releasedCount={Count}",
  988. tenantId, factoryId, affected);
  989. }
  990. return affected;
  991. }
  992. /// <summary>
  993. /// S8-SCHED-EXEC-1:到期规则候选 + 乐观 UPDATE 抢锁,返回成功抢到的 lease 列表。
  994. /// 抢锁条件:enabled=1 AND (paused_until IS NULL OR paused_until &lt;= NOW)
  995. /// AND (next_run_at IS NULL OR next_run_at &lt;= NOW)
  996. /// AND (lock_until IS NULL OR lock_until &lt;= NOW)。
  997. /// 抢锁回写:lock_token / locked_by / lock_until = NOW + 5min / running_started_at = NOW / last_run_id = runId。
  998. /// affectedRows == 1 才算抢到;后续 OnRuleCompletedAsync 必须按 lockToken 回写,避免旧进程覆盖新 lease。
  999. /// </summary>
  1000. public async Task<List<S8RuleLease>> PickReadyRulesAsync(long tenantId, long factoryId, int batchSize, string lockedBy, string runId)
  1001. {
  1002. if (batchSize <= 0) batchSize = 16;
  1003. var now = DateTime.Now;
  1004. var candidates = await _ruleRep.AsQueryable()
  1005. .Where(x => x.TenantId == tenantId
  1006. && x.FactoryId == factoryId
  1007. && x.Enabled
  1008. && (x.PausedUntil == null || x.PausedUntil <= now)
  1009. && (x.NextRunAt == null || x.NextRunAt <= now)
  1010. && (x.LockUntil == null || x.LockUntil <= now))
  1011. .OrderBy(x => x.NextRunAt, OrderByType.Asc)
  1012. .OrderBy(x => x.Id, OrderByType.Asc)
  1013. .Take(batchSize)
  1014. .Select(x => new { x.Id, x.RuleCode, x.RuleType })
  1015. .ToListAsync();
  1016. if (candidates.Count == 0) return new();
  1017. var leases = new List<S8RuleLease>();
  1018. foreach (var c in candidates)
  1019. {
  1020. var token = Guid.NewGuid().ToString("N");
  1021. var lockUntil = DateTime.Now.AddMinutes(LeaseDurationMinutes);
  1022. var runningAt = DateTime.Now;
  1023. // 乐观 UPDATE:再校验一次条件,affectedRows=1 才算抢到。
  1024. var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  1025. .SetColumns(x => new AdoS8WatchRule
  1026. {
  1027. LockToken = token,
  1028. LockedBy = lockedBy,
  1029. LockUntil = lockUntil,
  1030. RunningStartedAt = runningAt,
  1031. LastRunId = runId,
  1032. UpdatedAt = runningAt
  1033. })
  1034. .Where(x => x.Id == c.Id
  1035. && x.Enabled
  1036. && (x.PausedUntil == null || x.PausedUntil <= runningAt)
  1037. && (x.NextRunAt == null || x.NextRunAt <= runningAt)
  1038. && (x.LockUntil == null || x.LockUntil <= runningAt))
  1039. .ExecuteCommandAsync();
  1040. if (affected == 1)
  1041. {
  1042. leases.Add(new S8RuleLease
  1043. {
  1044. RuleId = c.Id,
  1045. RuleCode = c.RuleCode,
  1046. RuleType = c.RuleType,
  1047. LockToken = token,
  1048. LockedBy = lockedBy,
  1049. LockUntil = lockUntil,
  1050. RunId = runId,
  1051. AcquiredAt = runningAt
  1052. });
  1053. }
  1054. }
  1055. return leases;
  1056. }
  1057. /// <summary>
  1058. /// S8-SCHED-EXEC-1:执行单条已抢锁规则的 evaluator → 抗抖去重 → 建单/刷新 → 恢复 reconcile。
  1059. /// 不释放 lease(OnRuleCompletedAsync 负责);evaluator 抛异常时 Result.Success=false 并保留 ErrorMessage。
  1060. /// </summary>
  1061. public async Task<S8RuleRunResult> RunSingleRuleAsync(long tenantId, long factoryId, S8RuleLease lease)
  1062. {
  1063. var rule = await _ruleRep.AsQueryable()
  1064. .Where(x => x.Id == lease.RuleId)
  1065. .FirstAsync();
  1066. if (rule == null)
  1067. {
  1068. return new S8RuleRunResult { Success = false, ErrorMessage = "rule_not_found", Stats = new() };
  1069. }
  1070. var ruleType = rule.RuleType;
  1071. if (string.IsNullOrWhiteSpace(ruleType))
  1072. {
  1073. // 未分类的旧规则不在新调度路径承载;标 SKIPPED 但不视为失败。
  1074. return new S8RuleRunResult { Success = true, ErrorMessage = "rule_type_empty_skipped", Stats = new() };
  1075. }
  1076. IS8RuleEvaluator? evaluator = ruleType switch
  1077. {
  1078. S8TimeoutRuleEvaluator.RuleTypeCode => _timeoutEvaluator,
  1079. S8ShortageRuleEvaluator.RuleTypeCode => _shortageEvaluator,
  1080. S8OutOfRangeRuleEvaluator.RuleTypeCode => _outOfRangeEvaluator,
  1081. _ => null
  1082. };
  1083. if (evaluator == null)
  1084. {
  1085. return new S8RuleRunResult { Success = false, ErrorMessage = $"unsupported_rule_type:{ruleType}", Stats = new() };
  1086. }
  1087. try
  1088. {
  1089. var alertRules = (await _alertRuleRep.AsQueryable()
  1090. .Where(x => x.TenantId == tenantId && x.FactoryId == factoryId)
  1091. .ToListAsync()).AsReadOnly();
  1092. var results = await ProcessSingleRuleAsync(tenantId, factoryId, rule, ruleType, evaluator, alertRules, lease.RunId);
  1093. var stats = new S8RuleRunStats
  1094. {
  1095. Hits = results.Count,
  1096. Created = results.Count(r => r.Created),
  1097. Refreshed = results.Count(r => r.Reason == "duplicate_pending"),
  1098. Pending = results.Count(r => r.Reason == "antiflap_pending_hit"),
  1099. Failed = results.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
  1100. };
  1101. return new S8RuleRunResult { Success = true, Stats = stats };
  1102. }
  1103. catch (Exception ex)
  1104. {
  1105. return new S8RuleRunResult
  1106. {
  1107. Success = false,
  1108. ErrorMessage = ex.Message,
  1109. Stats = new()
  1110. };
  1111. }
  1112. }
  1113. /// <summary>
  1114. /// S8-SCHED-EXEC-1:单规则处理(evaluator → reconcile → hit 循环)。
  1115. /// 与 ProcessRulesByTypeAsync 内单规则循环体语义一致;此处抽出便于新调度路径直接调用单条 rule。
  1116. /// </summary>
  1117. private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync(
  1118. long tenantId, long factoryId, AdoS8WatchRule rule, string ruleType,
  1119. IS8RuleEvaluator evaluator, IReadOnlyList<AdoS8AlertRule> alertRules, string runId)
  1120. {
  1121. var results = new List<S8WatchCreationResult>();
  1122. List<S8RuleHit> hits;
  1123. try
  1124. {
  1125. hits = await evaluator.EvaluateAsync(tenantId, factoryId, rule, alertRules);
  1126. }
  1127. catch (Exception ex)
  1128. {
  1129. var failureReason = ex is S8RuleEvaluatorException sre ? sre.Reason : ex.GetType().Name;
  1130. await WriteDetectionLogAsync(new AdoS8DetectionLog
  1131. {
  1132. TenantId = tenantId, FactoryId = factoryId,
  1133. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  1134. SourceObjectType = rule.SourceObjectType,
  1135. DetectResult = DetectResultEvaluateFailed,
  1136. DetectedAt = DateTime.Now,
  1137. FailureReason = failureReason,
  1138. FailureMessage = Truncate(ex.Message, 1000),
  1139. RunId = runId, TriggerSource = DetectionTriggerSource
  1140. });
  1141. results.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
  1142. throw;
  1143. }
  1144. List<long> recoveredIds;
  1145. try
  1146. {
  1147. recoveredIds = await ReconcileRecoveriesForRuleAsync(tenantId, factoryId, rule, ruleType, hits, runId);
  1148. }
  1149. catch (Exception ex)
  1150. {
  1151. _logger.LogWarning(ex, "recovery_reconcile_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
  1152. recoveredIds = new();
  1153. }
  1154. if (hits.Count == 0 && recoveredIds.Count == 0)
  1155. {
  1156. await WriteDetectionLogAsync(new AdoS8DetectionLog
  1157. {
  1158. TenantId = tenantId, FactoryId = factoryId,
  1159. RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
  1160. SourceObjectType = rule.SourceObjectType,
  1161. DetectResult = DetectResultNoHit,
  1162. DetectedAt = DateTime.Now,
  1163. RunId = runId, TriggerSource = DetectionTriggerSource
  1164. });
  1165. }
  1166. foreach (var hit in hits)
  1167. {
  1168. if (string.IsNullOrWhiteSpace(hit.DedupKey))
  1169. {
  1170. results.Add(BuildSkipResult(rule, "missing_dedup_key", null, hit));
  1171. continue;
  1172. }
  1173. long matchedId;
  1174. try
  1175. {
  1176. matchedId = await FindOpenExceptionByDedupKeyAsync(tenantId, factoryId, hit.DedupKey);
  1177. }
  1178. catch (Exception ex)
  1179. {
  1180. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  1181. continue;
  1182. }
  1183. if (matchedId > 0)
  1184. {
  1185. try
  1186. {
  1187. await RefreshDetectionAsync(matchedId, hit);
  1188. await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultRefreshed, matchedId, runId));
  1189. results.Add(BuildSkippedDuplicate(rule, hit, matchedId));
  1190. }
  1191. catch (Exception ex)
  1192. {
  1193. results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
  1194. }
  1195. continue;
  1196. }
  1197. if (string.Equals(ruleType, S8OutOfRangeRuleEvaluator.RuleTypeCode, StringComparison.OrdinalIgnoreCase))
  1198. {
  1199. long compatId;
  1200. try
  1201. {
  1202. compatId = await FindLegacyOutOfRangeExceptionAsync(tenantId, factoryId, rule.Id, hit.RelatedObjectCode);
  1203. }
  1204. catch (Exception ex)
  1205. {
  1206. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  1207. continue;
  1208. }
  1209. if (compatId > 0)
  1210. {
  1211. try
  1212. {
  1213. await BackfillLegacyExceptionAsync(compatId, hit);
  1214. await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultRefreshed, compatId, runId));
  1215. results.Add(BuildSkippedDuplicate(rule, hit, compatId));
  1216. }
  1217. catch (Exception ex)
  1218. {
  1219. results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
  1220. }
  1221. continue;
  1222. }
  1223. }
  1224. bool typeExists;
  1225. try
  1226. {
  1227. typeExists = await _exceptionTypeRep.AsQueryable()
  1228. .Where(t => t.TypeCode == hit.ExceptionTypeCode
  1229. && (t.TenantId == 0 || t.TenantId == tenantId)
  1230. && (t.FactoryId == 0 || t.FactoryId == factoryId)
  1231. && t.Enabled)
  1232. .AnyAsync();
  1233. }
  1234. catch (Exception ex)
  1235. {
  1236. results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
  1237. continue;
  1238. }
  1239. if (!typeExists)
  1240. {
  1241. results.Add(BuildSkipResult(rule, "exception_type_missing", null, hit));
  1242. continue;
  1243. }
  1244. int hitCount;
  1245. AdoS8RuleDetectionState? state;
  1246. try
  1247. {
  1248. (state, hitCount) = await UpsertDetectionStateOnHitAsync(tenantId, factoryId, rule, hit);
  1249. }
  1250. catch (Exception ex)
  1251. {
  1252. results.Add(BuildSkipResult(rule, "antiflap_failed", ex.Message, hit));
  1253. continue;
  1254. }
  1255. var triggerRequired = NormalizeAntiflapCount(rule.TriggerCountRequired);
  1256. if (hitCount < triggerRequired)
  1257. {
  1258. _logger.LogInformation(
  1259. "antiflap_pending_hit ruleCode={RuleCode} dedupKey={DedupKey} hitCount={HitCount} trigger={Trigger}",
  1260. rule.RuleCode, hit.DedupKey, hitCount, triggerRequired);
  1261. results.Add(BuildSkipResult(rule, "antiflap_pending_hit", null, hit));
  1262. continue;
  1263. }
  1264. try
  1265. {
  1266. var entity = await _manualReportService.CreateFromHitAsync(hit);
  1267. await _exceptionRep.Context.Updateable<AdoS8Exception>()
  1268. .SetColumns(x => new AdoS8Exception
  1269. {
  1270. ConsecutiveHitCount = hitCount,
  1271. ConsecutiveMissCount = 0,
  1272. UpdatedAt = DateTime.Now
  1273. })
  1274. .Where(x => x.Id == entity.Id)
  1275. .ExecuteCommandAsync();
  1276. if (state != null)
  1277. {
  1278. await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
  1279. .SetColumns(x => new AdoS8RuleDetectionState
  1280. {
  1281. ActiveExceptionId = entity.Id,
  1282. UpdatedAt = DateTime.Now
  1283. })
  1284. .Where(x => x.Id == state.Id)
  1285. .ExecuteCommandAsync();
  1286. }
  1287. await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultCreated, entity.Id, runId));
  1288. results.Add(BuildCreatedResult(rule, hit, entity.Id));
  1289. }
  1290. catch (Exception ex)
  1291. {
  1292. results.Add(BuildSkipResult(rule, "create_failed", ex.Message, hit));
  1293. }
  1294. }
  1295. return results;
  1296. }
  1297. /// <summary>
  1298. /// S8-SCHED-EXEC-1:lease 执行完成回写。
  1299. /// 必须 WHERE id = lease.RuleId AND lock_token = lease.LockToken;affectedRows = 0 视为 lease 丢失,记录 Warning,不覆盖状态。
  1300. /// 失败 ≥ 阈值(默认 3)写 paused_until = NOW + 1h。
  1301. /// </summary>
  1302. public async Task OnRuleCompletedAsync(long tenantId, long factoryId, S8RuleLease lease, S8RuleRunResult result, int durationMs)
  1303. {
  1304. var rule = await _ruleRep.AsQueryable().Where(x => x.Id == lease.RuleId).FirstAsync();
  1305. if (rule == null)
  1306. {
  1307. _logger.LogWarning("lease_complete_rule_missing ruleId={RuleId}", lease.RuleId);
  1308. return;
  1309. }
  1310. var now = DateTime.Now;
  1311. var effectiveInterval = NormalizePollInterval(rule.PollIntervalSeconds);
  1312. var nextRunAt = now.AddSeconds(effectiveInterval);
  1313. int affected;
  1314. if (result.Success)
  1315. {
  1316. affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  1317. .SetColumns(x => new AdoS8WatchRule
  1318. {
  1319. LastRunAt = now,
  1320. NextRunAt = nextRunAt,
  1321. LastStatus = "SUCCESS",
  1322. LastError = null,
  1323. LastDurationMs = durationMs,
  1324. ConsecutiveFailureCount = 0,
  1325. LockToken = null,
  1326. LockedBy = null,
  1327. LockUntil = null,
  1328. RunningStartedAt = null,
  1329. UpdatedAt = now
  1330. })
  1331. .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
  1332. .ExecuteCommandAsync();
  1333. }
  1334. else
  1335. {
  1336. var errorTrunc = Truncate(result.ErrorMessage, 500);
  1337. var newFailures = rule.ConsecutiveFailureCount + 1;
  1338. DateTime? pausedUntil = rule.PausedUntil;
  1339. string? pauseReason = rule.PauseReason;
  1340. if (newFailures >= AutoPauseFailureThreshold)
  1341. {
  1342. pausedUntil = now.AddHours(AutoPauseDurationHours);
  1343. pauseReason = Truncate($"AUTO_PAUSED_AFTER_{AutoPauseFailureThreshold}_FAILURES: {errorTrunc}", 64);
  1344. }
  1345. affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
  1346. .SetColumns(x => new AdoS8WatchRule
  1347. {
  1348. LastRunAt = now,
  1349. NextRunAt = nextRunAt,
  1350. LastStatus = "FAILED",
  1351. LastError = errorTrunc,
  1352. LastDurationMs = durationMs,
  1353. ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
  1354. PausedUntil = pausedUntil,
  1355. PauseReason = pauseReason,
  1356. LockToken = null,
  1357. LockedBy = null,
  1358. LockUntil = null,
  1359. RunningStartedAt = null,
  1360. UpdatedAt = now
  1361. })
  1362. .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
  1363. .ExecuteCommandAsync();
  1364. }
  1365. if (affected == 0)
  1366. {
  1367. _logger.LogWarning(
  1368. "lease_lost_on_complete ruleId={RuleId} ruleCode={RuleCode} lockToken={LockToken}",
  1369. lease.RuleId, lease.RuleCode, lease.LockToken);
  1370. }
  1371. }
  1372. /// <summary>
  1373. /// S8-SCHED-EXEC-1:单 tick 完整流程。Job / debug 调度入口。
  1374. /// 1) ResetExpiredLeasesAsync
  1375. /// 2) PickReadyRulesAsync(batchSize)
  1376. /// 3) 每条 rule 独立 try/catch 调用 RunSingleRuleAsync + OnRuleCompletedAsync
  1377. /// 单条规则失败不影响其他规则;整 tick 不抛异常。
  1378. /// </summary>
  1379. public async Task<S8DispatchTickResult> RunDispatchTickAsync(long tenantId, long factoryId, int batchSize, string lockedBy)
  1380. {
  1381. var tickId = Guid.NewGuid().ToString("N").Substring(0, 8);
  1382. var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
  1383. var summary = new S8DispatchTickResult { TickId = tickId, RunId = runId };
  1384. try
  1385. {
  1386. summary.LeaseReleased = await ResetExpiredLeasesAsync(tenantId, factoryId);
  1387. }
  1388. catch (Exception ex)
  1389. {
  1390. _logger.LogError(ex, "tick_reset_lease_failed tickId={TickId}", tickId);
  1391. }
  1392. List<S8RuleLease> leases;
  1393. try
  1394. {
  1395. leases = await PickReadyRulesAsync(tenantId, factoryId, batchSize, lockedBy, runId);
  1396. }
  1397. catch (Exception ex)
  1398. {
  1399. _logger.LogError(ex, "tick_pick_failed tickId={TickId}", tickId);
  1400. return summary;
  1401. }
  1402. summary.Picked = leases.Count;
  1403. foreach (var lease in leases)
  1404. {
  1405. var sw = System.Diagnostics.Stopwatch.StartNew();
  1406. S8RuleRunResult runResult;
  1407. try
  1408. {
  1409. runResult = await RunSingleRuleAsync(tenantId, factoryId, lease);
  1410. }
  1411. catch (Exception ex)
  1412. {
  1413. runResult = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
  1414. }
  1415. sw.Stop();
  1416. try
  1417. {
  1418. await OnRuleCompletedAsync(tenantId, factoryId, lease, runResult, (int)sw.ElapsedMilliseconds);
  1419. }
  1420. catch (Exception ex)
  1421. {
  1422. _logger.LogError(ex, "tick_complete_failed tickId={TickId} ruleId={RuleId} ruleCode={RuleCode}", tickId, lease.RuleId, lease.RuleCode);
  1423. }
  1424. _logger.LogInformation(
  1425. "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}",
  1426. tickId, runId, lease.RuleId, lease.RuleCode,
  1427. runResult.Success ? "SUCCESS" : "FAILED",
  1428. sw.ElapsedMilliseconds,
  1429. runResult.Stats.Hits, runResult.Stats.Created, runResult.Stats.Refreshed, runResult.Stats.Pending, runResult.Stats.Failed,
  1430. runResult.ErrorMessage);
  1431. if (runResult.Success)
  1432. {
  1433. summary.Success++;
  1434. summary.Created += runResult.Stats.Created;
  1435. summary.Refreshed += runResult.Stats.Refreshed;
  1436. summary.Pending += runResult.Stats.Pending;
  1437. summary.PerRuleFailed += runResult.Stats.Failed;
  1438. }
  1439. else
  1440. {
  1441. summary.Failed++;
  1442. }
  1443. }
  1444. return summary;
  1445. }
  1446. private static int NormalizePollInterval(int raw)
  1447. {
  1448. if (raw < MinPollIntervalSeconds || raw > MaxPollIntervalSeconds) return DefaultPollIntervalSeconds;
  1449. return raw;
  1450. }
  1451. }
  1452. public sealed class S8WatchExecutionRule
  1453. {
  1454. public long WatchRuleId { get; set; }
  1455. public string WatchRuleCode { get; set; } = string.Empty;
  1456. public string SceneCode { get; set; } = string.Empty;
  1457. public string TriggerType { get; set; } = string.Empty;
  1458. public string WatchObjectType { get; set; } = string.Empty;
  1459. public long DataSourceId { get; set; }
  1460. public string DataSourceCode { get; set; } = string.Empty;
  1461. public string DataSourceType { get; set; } = string.Empty;
  1462. public string DataSourceConnection { get; set; } = string.Empty;
  1463. public string QueryExpression { get; set; } = string.Empty;
  1464. public int PollIntervalSeconds { get; set; }
  1465. public long AlertRuleId { get; set; }
  1466. public string AlertRuleCode { get; set; } = string.Empty;
  1467. public string TriggerCondition { get; set; } = string.Empty;
  1468. public string ThresholdValue { get; set; } = string.Empty;
  1469. public string Severity { get; set; } = string.Empty;
  1470. }
  1471. public sealed class S8WatchDeviceQueryResult
  1472. {
  1473. public long WatchRuleId { get; set; }
  1474. public string WatchRuleCode { get; set; } = string.Empty;
  1475. public bool Success { get; set; }
  1476. public string? FailureReason { get; set; }
  1477. public List<S8WatchDeviceRow> Rows { get; set; } = new();
  1478. public static S8WatchDeviceQueryResult Ok(S8WatchExecutionRule rule, List<S8WatchDeviceRow> rows) =>
  1479. new()
  1480. {
  1481. WatchRuleId = rule.WatchRuleId,
  1482. WatchRuleCode = rule.WatchRuleCode,
  1483. Success = true,
  1484. Rows = rows
  1485. };
  1486. public static S8WatchDeviceQueryResult Fail(S8WatchExecutionRule rule, string reason) =>
  1487. new()
  1488. {
  1489. WatchRuleId = rule.WatchRuleId,
  1490. WatchRuleCode = rule.WatchRuleCode,
  1491. Success = false,
  1492. FailureReason = reason
  1493. };
  1494. }
  1495. public sealed class S8WatchDeviceRow
  1496. {
  1497. public string RelatedObjectCode { get; set; } = string.Empty;
  1498. public decimal? CurrentValue { get; set; }
  1499. public long? OccurrenceDeptId { get; set; }
  1500. public long? ResponsibleDeptId { get; set; }
  1501. public string SourcePayload { get; set; } = string.Empty;
  1502. }
  1503. /// <summary>
  1504. /// G01-05 去重结果对象。仅服务 G01-06 建单前拦截,由 CanCreate 单决策位决定是否建单。
  1505. /// 只服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1506. /// 不预留多 trigger_type / 平台化去重扩展结构。
  1507. /// Reason 值域:no_pending / duplicate_pending / missing_dedup_key / query_failed。
  1508. /// </summary>
  1509. public sealed class S8WatchDedupResult
  1510. {
  1511. public S8WatchHitResult Hit { get; set; } = new();
  1512. public bool CanCreate { get; set; }
  1513. public long? MatchedExceptionId { get; set; }
  1514. public string Reason { get; set; } = string.Empty;
  1515. }
  1516. /// <summary>
  1517. /// G01-06 建单结果对象。仅服务 G-01 首版主线验收,由 Created / Skipped 两位决定结局。
  1518. /// 只服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1519. /// 不预留多 trigger_type / 平台化工单扩展结构。
  1520. /// Reason 值域:auto_created / create_failed / 透传自 DedupResult.Reason。
  1521. /// </summary>
  1522. public sealed class S8WatchCreationResult
  1523. {
  1524. public S8WatchDedupResult DedupResult { get; set; } = new();
  1525. public bool Created { get; set; }
  1526. public bool Skipped { get; set; }
  1527. public long? CreatedExceptionId { get; set; }
  1528. public string Reason { get; set; } = string.Empty;
  1529. public string? ErrorMessage { get; set; }
  1530. }
  1531. /// <summary>
  1532. /// G01-04 命中结果对象。承载 G01-05 去重与 G01-06 建单所需最小追溯字段,
  1533. /// 仅服务首版唯一场景 S2S6_PRODUCTION + 唯一 trigger_type VALUE_DEVIATION + 设备对象。
  1534. /// 不预留多 trigger_type / 多场景 / 平台化扩展结构。
  1535. /// </summary>
  1536. public sealed class S8WatchHitResult
  1537. {
  1538. public long SourceRuleId { get; set; }
  1539. public string SourceRuleCode { get; set; } = string.Empty;
  1540. public long AlertRuleId { get; set; }
  1541. public long DataSourceId { get; set; }
  1542. public string RelatedObjectCode { get; set; } = string.Empty;
  1543. public decimal CurrentValue { get; set; }
  1544. public decimal ThresholdValue { get; set; }
  1545. public string TriggerCondition { get; set; } = string.Empty;
  1546. public string Severity { get; set; } = string.Empty;
  1547. public long? OccurrenceDeptId { get; set; }
  1548. public long? ResponsibleDeptId { get; set; }
  1549. public string SourcePayload { get; set; } = string.Empty;
  1550. }
  1551. /// <summary>S8-SCHED-EXEC-1:lease 抢占成功后传递的最小标识对象。</summary>
  1552. public sealed class S8RuleLease
  1553. {
  1554. public long RuleId { get; set; }
  1555. public string RuleCode { get; set; } = string.Empty;
  1556. public string? RuleType { get; set; }
  1557. public string LockToken { get; set; } = string.Empty;
  1558. public string LockedBy { get; set; } = string.Empty;
  1559. public DateTime LockUntil { get; set; }
  1560. public string RunId { get; set; } = string.Empty;
  1561. public DateTime AcquiredAt { get; set; }
  1562. }
  1563. /// <summary>S8-SCHED-EXEC-1:单条规则执行结果,OnRuleCompletedAsync 据此更新状态。</summary>
  1564. public sealed class S8RuleRunResult
  1565. {
  1566. public bool Success { get; set; }
  1567. public string? ErrorMessage { get; set; }
  1568. public S8RuleRunStats Stats { get; set; } = new();
  1569. }
  1570. public sealed class S8RuleRunStats
  1571. {
  1572. public int Hits { get; set; }
  1573. public int Created { get; set; }
  1574. public int Refreshed { get; set; }
  1575. public int Pending { get; set; }
  1576. public int Failed { get; set; }
  1577. }
  1578. /// <summary>S8-SCHED-EXEC-1:单 tick 调度结果聚合。</summary>
  1579. public sealed class S8DispatchTickResult
  1580. {
  1581. public string TickId { get; set; } = string.Empty;
  1582. public string RunId { get; set; } = string.Empty;
  1583. public int LeaseReleased { get; set; }
  1584. public int Picked { get; set; }
  1585. public int Success { get; set; }
  1586. public int Failed { get; set; }
  1587. public int Created { get; set; }
  1588. public int Refreshed { get; set; }
  1589. public int Pending { get; set; }
  1590. public int PerRuleFailed { get; set; }
  1591. }