diff --git a/src/TrackCenter.Repository/ES/EsExtensions.cs b/src/TrackCenter.Repository/ES/EsExtensions.cs index 2b0e1578..faed3b8e 100644 --- a/src/TrackCenter.Repository/ES/EsExtensions.cs +++ b/src/TrackCenter.Repository/ES/EsExtensions.cs @@ -9,38 +9,61 @@ namespace TrackCenter.Repository.ES; /// public static class EsExtensions { + /// + /// 月度索引读请求：走 {index}-* 通配符，无匹配索引时返回空结果（首次部署 / 尚未创建别名）。 + /// + public static SearchDescriptor UseMonthlyIndex(this SearchDescriptor descriptor, string indexPrefix) + where T : class + { + return descriptor + .Index($"{indexPrefix}-*") + .IgnoreUnavailable() + .AllowNoIndices(); + } + + /// + /// 月度索引按查询更新：目标索引不存在时视为无匹配文档。 + /// + public static UpdateByQueryDescriptor UseMonthlyIndex(this UpdateByQueryDescriptor descriptor, + string indexPrefix) + where T : class + { + return descriptor + .Index($"{indexPrefix}-*") + .IgnoreUnavailable() + .AllowNoIndices(); + } + + /// + /// 月度索引按查询删除：目标索引不存在时视为无匹配文档。 + /// + public static DeleteByQueryDescriptor UseMonthlyIndex(this DeleteByQueryDescriptor descriptor, + string indexPrefix) + where T : class + { + return descriptor + .Index($"{indexPrefix}-*") + .IgnoreUnavailable() + .AllowNoIndices(); + } + /// /// 轨迹明细实际写入月度索引 track_details-yyyy.MM；读请求走通配符， /// 无匹配索引时返回空结果（首次部署 / 尚未 Migrate 别名）。 /// public static SearchDescriptor UseTrackDetailsIndex(this SearchDescriptor descriptor) where T : class - { - return descriptor - .Index($"{GlobalConst.TrackDetailsIndex}-*") - .IgnoreUnavailable() - .AllowNoIndices(); - } + => descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex); /// /// 轨迹明细按查询更新：目标索引不存在时视为无匹配文档。 /// public static UpdateByQueryDescriptor UseTrackDetailsIndex(this UpdateByQueryDescriptor descriptor) where T : class - { - return descriptor - .Index($"{GlobalConst.TrackDetailsIndex}-*") - .IgnoreUnavailable() - .AllowNoIndices(); - } + => descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex); public static DeleteByQueryDescriptor UseTrackDetailsIndex(this DeleteByQueryDescriptor descriptor) where T : class - { - return descriptor - .Index($"{GlobalConst.TrackDetailsIndex}-*") - .IgnoreUnavailable() - .AllowNoIndices(); - } + => descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex); /// /// 索引/别名不存在（含 NEST 对 404 的包装错误）。 diff --git a/src/TrackCenter.Repository/Repositories/OrderOperationRecordEsRepository.cs b/src/TrackCenter.Repository/Repositories/OrderOperationRecordEsRepository.cs index 50332507..751e6102 100644 --- a/src/TrackCenter.Repository/Repositories/OrderOperationRecordEsRepository.cs +++ b/src/TrackCenter.Repository/Repositories/OrderOperationRecordEsRepository.cs @@ -67,13 +67,21 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract> GetOrderOperationsByIdAsync(int orderId) { var data = await Context.SearchAsync(x => x - .Index(GlobalConst.OrderOperationRecordIndex) + .UseMonthlyIndex(GlobalConst.OrderOperationRecordIndex) .Query(q => q.Term(t => t.OrderId, orderId) ) .Size(GlobalConst.NormalLimit) .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending))); + if (!data.IsValid) + { + if (data.IsIndexUnavailable()) + return []; + + throw new InvalidOperationException($"订单操作记录查询失败，原因：{data.DebugInformation}。"); + } + return data.ToIdList(); } } diff --git a/src/TrackCenter.Repository/Repositories/TrackPushRecordEsRepository.cs b/src/TrackCenter.Repository/Repositories/TrackPushRecordEsRepository.cs index 1424e571..2b6cf7b6 100644 --- a/src/TrackCenter.Repository/Repositories/TrackPushRecordEsRepository.cs +++ b/src/TrackCenter.Repository/Repositories/TrackPushRecordEsRepository.cs @@ -69,13 +69,21 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract> GetTrackPushRecordsAsync(int orderId) { var data = await Context.SearchAsync(x => x - .Index(GlobalConst.TrackPushRecordIndex) + .UseMonthlyIndex(GlobalConst.TrackPushRecordIndex) .Query(q => q.Term(t => t.OrderId, orderId) ) .Size(1000) .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending))); + if (!data.IsValid) + { + if (data.IsIndexUnavailable()) + return []; + + throw new InvalidOperationException($"轨迹推送记录查询失败，原因：{data.DebugInformation}。"); + } + return data.ToIdList(); } @@ -114,7 +122,7 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract(x => x - .Index(GlobalConst.TrackPushRecordIndex) + .UseMonthlyIndex(GlobalConst.TrackPushRecordIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.IsSuccess).Terms(input.IsSuccess))) .Should(musts) @@ -126,7 +134,7 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract mu.Terms(t => t.Field(f => f.IsSuccess).Terms(input.IsSuccess))); r = await Context.SearchAsync(x => x - .Index(GlobalConst.TrackPushRecordIndex) + .UseMonthlyIndex(GlobalConst.TrackPushRecordIndex) .Query(q => q.Bool(b => b .Must(musts)) ) @@ -136,6 +144,9 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract mu.Term(t => t.IsSuccess, true)); var data = await Context.SearchAsync(x => x - .Index(GlobalConst.TrackPushRecordIndex) + .UseMonthlyIndex(GlobalConst.TrackPushRecordIndex) .Query(q => q.Bool(b => b.Must(musts)) @@ -161,13 +172,21 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); + if (!data.IsValid) + { + if (data.IsIndexUnavailable()) + return null; + + throw new InvalidOperationException($"轨迹推送记录查询失败，原因：{data.DebugInformation}。"); + } + return data.Documents.FirstOrDefault(); } public async Task> GetHistorySuccessTrackPushRecordAsync(int orderId) { var data = await Context.SearchAsync(x => x - .Index(GlobalConst.TrackPushRecordIndex) + .UseMonthlyIndex(GlobalConst.TrackPushRecordIndex) .Source(s => s.Excludes(fs => fs.Field(f => f.PushContent))) @@ -179,6 +198,15 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); + + if (!data.IsValid) + { + if (data.IsIndexUnavailable()) + return []; + + throw new InvalidOperationException($"轨迹推送记录查询失败，原因：{data.DebugInformation}。"); + } + return data.ToIdList(); } } diff --git a/src/TrackCenter.Repository/Repositories/WarningDetailsEsRepository.cs b/src/TrackCenter.Repository/Repositories/WarningDetailsEsRepository.cs index fcb6b327..415d65ae 100644 --- a/src/TrackCenter.Repository/Repositories/WarningDetailsEsRepository.cs +++ b/src/TrackCenter.Repository/Repositories/WarningDetailsEsRepository.cs @@ -50,7 +50,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien c => c.Match(f => f.Field(d => d.NotifyUsers).Query(input.NotifyUserId))); var r = await _client.SearchAsync(x => x - .Index(GlobalConst.TransportWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) @@ -60,6 +60,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return new PagedResult(0, []); + throw new InvalidOperationException($"运输预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -116,7 +119,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (isShouldQuery) { r = await _client.SearchAsync(x => x - .Index(GlobalConst.TransportWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) @@ -128,7 +131,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien { musts.Add(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x - .Index(GlobalConst.TransportWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Size(SafelyLimit) ); @@ -136,6 +139,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return []; + throw new InvalidOperationException($"运输预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -294,7 +300,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => - u.Index(GlobalConst.TransportWarningDetailsIndex) + u.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source(@"ctx._source.status = params.status; @@ -310,7 +316,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien ) ); - if (!r.IsValid) + if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"运输时效预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。"); @@ -387,7 +393,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien c => c.Match(f => f.Field(d => d.Customer).Query(input.Customer))); var r = await _client.SearchAsync(x => x - .Index(GlobalConst.ReturnWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) @@ -397,6 +403,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return new PagedResult(0, []); + throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -442,7 +451,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (isShouldQuery) { r = await _client.SearchAsync(x => x - .Index(GlobalConst.ReturnWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) @@ -454,7 +463,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien { musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x - .Index(GlobalConst.ReturnWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(musts) )) @@ -464,6 +473,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return []; + throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -473,13 +485,16 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien public async Task> GetReturnMessageRecordListByIdAsync(string esId) { var r = await _client.SearchAsync(x => x - .Index(GlobalConst.ReturnMessageRecordIndex) + .UseMonthlyIndex(GlobalConst.ReturnMessageRecordIndex) .Query(q => q.Match(f => f.Field(d => d.WarningDetailsIds).Query(esId))) .Size(SafelyLimit) ); if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return []; + throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -489,13 +504,13 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien public async Task ReturnWarningDetailsRetryIncrementAsync(HashSet esIds) { var r = await _client.UpdateByQueryAsync(u => - u.Index(GlobalConst.ReturnWarningDetailsIndex) + u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Ids(i => i.Values(esIds))) .Script(s => s .Source("ctx._source.retryCount = ctx._source.retryCount+1;") ) ); - if (!r.IsValid) + if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException($"回传预警明细更新失败，原因：{r.DebugInformation}。"); } @@ -661,7 +676,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => - u.Index(GlobalConst.ReturnWarningDetailsIndex) + u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source("ctx._source.status = params.status") @@ -669,7 +684,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien ) ); - if (!r.IsValid) + if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"回传预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。"); @@ -766,7 +781,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien .Params(p => p.Add("value", input.MissingTrackStatusId.ToString()))))); var r = await _client.SearchAsync(x => x - .Index(GlobalConst.CompleteWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Sort(s => s.Ascending(f => f.Status).Descending(f => f.CreateTime)) .From((input.PageIndex - 1) * input.PageSize) @@ -776,6 +791,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return new PagedResult(0, []); + throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -830,7 +848,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (isShouldQuery) { r = await _client.SearchAsync(x => x - .Index(GlobalConst.CompleteWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status))) .Should(musts) @@ -842,7 +860,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien { musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status))); r = await _client.SearchAsync(x => x - .Index(GlobalConst.CompleteWarningDetailsIndex) + .UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b .Must(musts) )) @@ -852,6 +870,9 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien if (!r.IsValid) { + if (r.IsIndexUnavailable()) + return []; + throw new InvalidOperationException($"完整预警明细查询失败，原因：{r.DebugInformation}。"); } @@ -872,7 +893,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien musts.AddIf(x => x.HasOrderIds, c => c.Terms(t => t.Field(f => f.OrderId).Terms(input.OrderIds))); var r = await _client.UpdateByQueryAsync(u => - u.Index(GlobalConst.CompleteWarningDetailsIndex) + u.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex) .Query(q => q.Bool(b => b.Must(musts))) .Script(s => s .Source(@" @@ -886,7 +907,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien ) ); - if (!r.IsValid) + if (!r.IsValid && !r.IsIndexUnavailable()) { throw new InvalidOperationException( $"完整预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。");