This commit is contained in:
parent
cdff57f8c3
commit
a39856db9c
@ -1,5 +1,5 @@
|
||||
using TrackCenter.Repository.ES;
|
||||
using Nest;
|
||||
using TrackCenter.Domain.Consts;
|
||||
using TrackCenter.Domain.SeedWork;
|
||||
|
||||
namespace TrackCenter.Repository.ES;
|
||||
@ -9,12 +9,74 @@ namespace TrackCenter.Repository.ES;
|
||||
/// </summary>
|
||||
public static class EsExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// 轨迹明细实际写入月度索引 track_details-yyyy.MM;读请求走通配符,
|
||||
/// 无匹配索引时返回空结果(首次部署 / 尚未 Migrate 别名)。
|
||||
/// </summary>
|
||||
public static SearchDescriptor<T> UseTrackDetailsIndex<T>(this SearchDescriptor<T> descriptor) where T : class
|
||||
{
|
||||
return descriptor
|
||||
.Index($"{GlobalConst.TrackDetailsIndex}-*")
|
||||
.IgnoreUnavailable()
|
||||
.AllowNoIndices();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹明细按查询更新:目标索引不存在时视为无匹配文档。
|
||||
/// </summary>
|
||||
public static UpdateByQueryDescriptor<T> UseTrackDetailsIndex<T>(this UpdateByQueryDescriptor<T> descriptor)
|
||||
where T : class
|
||||
{
|
||||
return descriptor
|
||||
.Index($"{GlobalConst.TrackDetailsIndex}-*")
|
||||
.IgnoreUnavailable()
|
||||
.AllowNoIndices();
|
||||
}
|
||||
|
||||
public static DeleteByQueryDescriptor<T> UseTrackDetailsIndex<T>(this DeleteByQueryDescriptor<T> descriptor)
|
||||
where T : class
|
||||
{
|
||||
return descriptor
|
||||
.Index($"{GlobalConst.TrackDetailsIndex}-*")
|
||||
.IgnoreUnavailable()
|
||||
.AllowNoIndices();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 索引/别名不存在(含 NEST 对 404 的包装错误)。
|
||||
/// </summary>
|
||||
public static bool IsIndexUnavailable(this IResponse response)
|
||||
{
|
||||
if (response.IsValid)
|
||||
return false;
|
||||
|
||||
if (response.ApiCall?.HttpStatusCode == 404)
|
||||
return true;
|
||||
|
||||
if (IsIndexNotFoundType(response.ServerError?.Error?.Type)
|
||||
|| IsIndexNotFoundType(response.ServerError?.Error?.CausedBy?.Type))
|
||||
return true;
|
||||
|
||||
var rootCauses = response.ServerError?.Error?.RootCause;
|
||||
if (rootCauses != null && rootCauses.Any(x => IsIndexNotFoundType(x.Type)))
|
||||
return true;
|
||||
|
||||
// NEST: Invalid NEST response built from a successful (404) low level call on POST: /track_details/_search
|
||||
var debug = response.DebugInformation;
|
||||
if (!string.IsNullOrEmpty(debug)
|
||||
&& (debug.Contains("index_not_found_exception", StringComparison.OrdinalIgnoreCase)
|
||||
|| debug.Contains("successful (404)", StringComparison.OrdinalIgnoreCase)))
|
||||
return true;
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
private static bool IsIndexNotFoundType(string? type) =>
|
||||
string.Equals(type, "index_not_found_exception", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>
|
||||
/// return list with id
|
||||
/// </summary>
|
||||
/// <param name="response"></param>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static List<T> ToIdList<T>(this ISearchResponse<T> response) where T : IdBaseIndex
|
||||
{
|
||||
var dtos = new List<T>();
|
||||
|
||||
@ -25,7 +25,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
{
|
||||
var r = forUpsert
|
||||
? await Context.SearchAsync<TrackDetailsIndex>(s => s
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Source()
|
||||
.Version()
|
||||
.Size(SafelyLimit)
|
||||
@ -39,7 +39,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
)
|
||||
)
|
||||
: await Context.SearchAsync<TrackDetailsIndex>(s => s
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Source(s2 => s2.Includes(fd => fd
|
||||
.Field(f => f.IndexName)
|
||||
@ -60,6 +60,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCode:{trackCode},原因:{r.DebugInformation}。");
|
||||
}
|
||||
@ -126,7 +129,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
if (isShouldQuery)
|
||||
{
|
||||
r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Query(q => q.Bool(b => b
|
||||
.Should(musts)
|
||||
@ -137,7 +140,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
else
|
||||
{
|
||||
r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Query(q => q.Bool(b => b
|
||||
.Must(musts)
|
||||
@ -148,6 +151,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new Dictionary<int, List<TrackDetailsIndex>>();
|
||||
|
||||
throw new InvalidOperationException($"运输预警明细查询失败,原因:{r.DebugInformation}。");
|
||||
}
|
||||
|
||||
@ -159,7 +165,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<Dictionary<int, Dictionary<int, TrackDetailsIndex>>> GetTrackStatusEarliestDictAsync(HashSet<int> orderIds)
|
||||
{
|
||||
var response = await Context.SearchAsync<TrackDetailsIndex>(s => s
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Size(0) // 不需要实际文档,只需要聚合结果
|
||||
.Query(q => q
|
||||
.Terms(t => t
|
||||
@ -191,6 +197,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!response.IsValid)
|
||||
{
|
||||
if (response.IsIndexUnavailable())
|
||||
return new Dictionary<int, Dictionary<int, TrackDetailsIndex>>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"最新轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],原因:{response.DebugInformation}。");
|
||||
}
|
||||
@ -226,7 +235,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<Dictionary<int, TrackDetailsIndex>> GetLatestDictAsync(HashSet<int> orderIds)
|
||||
{
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(s => s
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Size(0) // 不需要实际文档,只需要聚合结果
|
||||
.Query(q => q
|
||||
.Terms(t => t
|
||||
@ -253,6 +262,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new Dictionary<int, TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"最新轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],原因:{r.DebugInformation}。");
|
||||
}
|
||||
@ -281,7 +293,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
var excludeTrackCodes = trackCodes.ToList();
|
||||
excludeTrackCodes.Add(GlobalConst.EmptyString);
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Source(s => s.Includes(fd => fd
|
||||
.Field(f => f.TrackEnContent)
|
||||
.Field(f => f.TrackContent)
|
||||
@ -335,6 +347,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"最新轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCodes:[{trackCodes.JoinAsString(", ")}],原因:{r.DebugInformation}。");
|
||||
}
|
||||
@ -345,7 +360,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<bool> AnyAsync(int orderId, string originTrackDataHash)
|
||||
{
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Source(s => s.Includes(fd => fd.Field(f => f.TrackCode)))
|
||||
.Query(q =>
|
||||
q.Term(f => f.OrderId, orderId)
|
||||
@ -355,6 +370,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return false;
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"轨迹查询失败,orderId:{orderId},hash:{originTrackDataHash},原因:{r.DebugInformation}。");
|
||||
}
|
||||
@ -412,7 +430,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
while (retryCount < maxRetries && !updateSuccessful)
|
||||
{
|
||||
response = await Context.UpdateByQueryAsync<TrackDetailsIndex>(u => u
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Query(q => q
|
||||
.Ids(t => t.Values(indexes.Select(x => x.Id)))
|
||||
)
|
||||
@ -465,7 +483,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
)
|
||||
);
|
||||
|
||||
if (response.IsValid)
|
||||
if (response.IsValid || response.IsIndexUnavailable())
|
||||
{
|
||||
updateSuccessful = true;
|
||||
}
|
||||
@ -542,7 +560,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
var r = await PollyHelper.RetryPollyExecAsync(() => Context.BulkAsync(desc),
|
||||
successValidator: response =>
|
||||
{
|
||||
if (response.IsValid) // 认为成功
|
||||
if (response.IsValid || response.IsIndexUnavailable()) // 认为成功或索引不存在
|
||||
return true;
|
||||
|
||||
if (response.ServerError.Status == 409) //版本冲突认为失败,则重试
|
||||
@ -620,7 +638,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Source()
|
||||
.Version()
|
||||
.Query(q =>
|
||||
@ -637,6 +655,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"轨迹查询失败,orderType:{orderType},orderNos:{orderNos},原因:{r.DebugInformation}。");
|
||||
}
|
||||
@ -647,7 +668,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)
|
||||
{
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Query(q =>
|
||||
q.Term(t => t.OrderId, orderId)
|
||||
@ -660,6 +681,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException($"轨迹查询失败,orderId:{orderId},原因:{r.DebugInformation}。");
|
||||
}
|
||||
|
||||
@ -686,7 +710,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
}
|
||||
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Query(_ => query)
|
||||
.Size(SafelyLimit)
|
||||
@ -697,6 +721,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException($"轨迹查询失败,orderIds:[{string.Join(',', orderIds)}],原因:{r.DebugInformation}。");
|
||||
}
|
||||
|
||||
@ -720,7 +747,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
// }
|
||||
|
||||
var search = await Context.SearchAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Source(s => s.Includes(fd => fd
|
||||
.Field(f => f.IndexName)
|
||||
@ -736,6 +763,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!search.IsValid)
|
||||
{
|
||||
if (search.IsIndexUnavailable())
|
||||
return new List<int>();
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"轨迹查询失败,orderIds:[{orderIds.JoinAsString(", ")}],trackCodes:[{trackCodes.JoinAsString(", ")}],原因:{search.DebugInformation}。");
|
||||
}
|
||||
@ -774,12 +804,15 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task DeleteTrackByIdsAsync(HashSet<string> esIds)
|
||||
{
|
||||
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Query(q => q.Ids(i => i.Values(esIds)))
|
||||
);
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return;
|
||||
|
||||
throw new InvalidOperationException($"轨迹删除失败,ids:[{string.Join(", ", esIds)}],原因:{r.DebugInformation}。");
|
||||
}
|
||||
}
|
||||
@ -787,7 +820,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<List<TrackDetailsIndex>> GetDataByIds(string[] ids)
|
||||
{
|
||||
var r = await Context.SearchAsync<TrackDetailsIndex>(m => m
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Version()
|
||||
.Query(
|
||||
q => q.Ids(i => i.Values(ids))
|
||||
@ -795,6 +828,9 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
|
||||
if (!r.IsValid)
|
||||
{
|
||||
if (r.IsIndexUnavailable())
|
||||
return new List<TrackDetailsIndex>();
|
||||
|
||||
throw new InvalidOperationException($"轨迹查询失败,ids:[{string.Join(", ", ids)}],原因:{r.DebugInformation}。");
|
||||
}
|
||||
|
||||
@ -804,7 +840,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
public async Task<TrackDetailsIndex?> GetDetailAsync(string orderNo, string trackCode, TrackOrderType orderType, string appKey, DateTime trackTime)
|
||||
{
|
||||
var res = await Context.SearchAsync<TrackDetailsIndex>(s => s
|
||||
.Index(GlobalConst.TrackDetailsIndex)
|
||||
.UseTrackDetailsIndex()
|
||||
.Query(q => q
|
||||
.Bool(b => b
|
||||
.Must(mu => mu
|
||||
@ -832,6 +868,15 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
||||
)
|
||||
);
|
||||
|
||||
if (!res.IsValid)
|
||||
{
|
||||
if (res.IsIndexUnavailable())
|
||||
return null;
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"轨迹查询失败,orderNo:{orderNo},trackCode:{trackCode},原因:{res.DebugInformation}。");
|
||||
}
|
||||
|
||||
return res.Documents.FirstOrDefault();
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user