ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: ideship:4cae065e17f18407ec5489c927a8923e6b8e2dce
Branches Tags
ideship:master
ideship:test
..
compare: ideship:f6590de824204529267fef13cb9418cfc3f53e22
Branches Tags
ideship:master
ideship:test

No commits in common. "4cae065e17f18407ec5489c927a8923e6b8e2dce" and "f6590de824204529267fef13cb9418cfc3f53e22" have entirely different histories.
4cae065e17 ... f6590de824

4 changed files with 42 additions and 122 deletions
Show Stats Expand all files Collapse all files

59
src/TrackCenter.Repository/ES/EsExtensions.cs
View File

@ -9,61 +9,38 @@ namespace TrackCenter.Repository.ES;
/// </summary>
public static class EsExtensions
{
/// <summary>
/// 月度索引读请求：走 {index}-* 通配符，无匹配索引时返回空结果（首次部署 / 尚未创建别名）。
/// </summary>
public static SearchDescriptor<T> UseMonthlyIndex<T>(this SearchDescriptor<T> descriptor, string indexPrefix)
where T : class
{
return descriptor
.Index($"{indexPrefix}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
/// <summary>
/// 月度索引按查询更新：目标索引不存在时视为无匹配文档。
/// </summary>
public static UpdateByQueryDescriptor<T> UseMonthlyIndex<T>(this UpdateByQueryDescriptor<T> descriptor,
string indexPrefix)
where T : class
{
return descriptor
.Index($"{indexPrefix}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
/// <summary>
/// 月度索引按查询删除：目标索引不存在时视为无匹配文档。
/// </summary>
public static DeleteByQueryDescriptor<T> UseMonthlyIndex<T>(this DeleteByQueryDescriptor<T> descriptor,
string indexPrefix)
where T : class
{
return descriptor
.Index($"{indexPrefix}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
/// <summary>
/// 轨迹明细实际写入月度索引 track_details-yyyy.MM；读请求走通配符，
/// 无匹配索引时返回空结果（首次部署 / 尚未 Migrate 别名）。
/// </summary>
public static SearchDescriptor<T> UseTrackDetailsIndex<T>(this SearchDescriptor<T> descriptor) where T : class
=> descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex);
{
return descriptor
.Index($"{GlobalConst.TrackDetailsIndex}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
/// <summary>
/// 轨迹明细按查询更新：目标索引不存在时视为无匹配文档。
/// </summary>
public static UpdateByQueryDescriptor<T> UseTrackDetailsIndex<T>(this UpdateByQueryDescriptor<T> descriptor)
where T : class
=> descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex);
{
return descriptor
.Index($"{GlobalConst.TrackDetailsIndex}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
public static DeleteByQueryDescriptor<T> UseTrackDetailsIndex<T>(this DeleteByQueryDescriptor<T> descriptor)
where T : class
=> descriptor.UseMonthlyIndex(GlobalConst.TrackDetailsIndex);
{
return descriptor
.Index($"{GlobalConst.TrackDetailsIndex}-*")
.IgnoreUnavailable()
.AllowNoIndices();
}
/// <summary>
/// 索引/别名不存在（含 NEST 对 404 的包装错误）。

10
src/TrackCenter.Repository/Repositories/OrderOperationRecordEsRepository.cs
View File

@ -67,21 +67,13 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract<Ord
public async Task<List<OrderOperationRecordIndex>> GetOrderOperationsByIdAsync(int orderId)
{
var data = await Context.SearchAsync<OrderOperationRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.OrderOperationRecordIndex)
.Index(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();
}
}

38
src/TrackCenter.Repository/Repositories/TrackPushRecordEsRepository.cs
View File

@ -69,21 +69,13 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
public async Task<List<TrackPushRecordIndex>> GetTrackPushRecordsAsync(int orderId)
{
var data = await Context.SearchAsync<TrackPushRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.TrackPushRecordIndex)
.Index(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();
}
@ -122,7 +114,7 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
if (isShouldQuery)
{
r = await Context.SearchAsync<TrackPushRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.TrackPushRecordIndex)
.Index(GlobalConst.TrackPushRecordIndex)
.Query(q => q.Bool(b => b
.Must(mu => mu.Terms(t => t.Field(f => f.IsSuccess).Terms(input.IsSuccess)))
.Should(musts)
@ -134,7 +126,7 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
{
musts.Add(mu => mu.Terms(t => t.Field(f => f.IsSuccess).Terms(input.IsSuccess)));
r = await Context.SearchAsync<TrackPushRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.TrackPushRecordIndex)
.Index(GlobalConst.TrackPushRecordIndex)
.Query(q => q.Bool(b => b
.Must(musts))
)
@ -144,9 +136,6 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return [];
throw new InvalidOperationException($"轨迹推送记录查询失败，原因：{r.DebugInformation}。");
}
@ -164,7 +153,7 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
musts.Add(mu => mu.Term(t => t.IsSuccess, true));
var data = await Context.SearchAsync<TrackPushRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.TrackPushRecordIndex)
.Index(GlobalConst.TrackPushRecordIndex)
.Query(q =>
q.Bool(b =>
b.Must(musts))
@ -172,21 +161,13 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
.Size(1000)
.Sort(s => 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<List<TrackPushRecordIndex>> GetHistorySuccessTrackPushRecordAsync(int orderId)
{
var data = await Context.SearchAsync<TrackPushRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.TrackPushRecordIndex)
.Index(GlobalConst.TrackPushRecordIndex)
.Source(s =>
s.Excludes(fs =>
fs.Field(f => f.PushContent)))
@ -198,15 +179,6 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
)
.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();
}
}

57
src/TrackCenter.Repository/Repositories/WarningDetailsEsRepository.cs
View File

@ -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<TransportWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex)
.Index(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,9 +60,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return new PagedResult<TransportWarningDetailsIndex>(0, []);
throw new InvalidOperationException($"运输预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -119,7 +116,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (isShouldQuery)
{
r = await _client.SearchAsync<TransportWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex)
.Index(GlobalConst.TransportWarningDetailsIndex)
.Query(q => q.Bool(b => b
.Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status)))
.Should(musts)
@ -131,7 +128,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
{
musts.Add(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status)));
r = await _client.SearchAsync<TransportWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex)
.Index(GlobalConst.TransportWarningDetailsIndex)
.Query(q => q.Bool(b => b.Must(musts)))
.Size(SafelyLimit)
);
@ -139,9 +136,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return [];
throw new InvalidOperationException($"运输预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -300,7 +294,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<TransportWarningDetailsIndex>(u =>
u.UseMonthlyIndex(GlobalConst.TransportWarningDetailsIndex)
u.Index(GlobalConst.TransportWarningDetailsIndex)
.Query(q => q.Bool(b => b.Must(musts)))
.Script(s => s
.Source(@"ctx._source.status = params.status;
@ -316,7 +310,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
)
);
if (!r.IsValid && !r.IsIndexUnavailable())
if (!r.IsValid)
{
throw new InvalidOperationException(
$"运输时效预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。");
@ -393,7 +387,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
c => c.Match(f => f.Field(d => d.Customer).Query(input.Customer)));
var r = await _client.SearchAsync<ReturnWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex)
.Index(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)
@ -403,9 +397,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return new PagedResult<ReturnWarningDetailsIndex>(0, []);
throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -451,7 +442,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (isShouldQuery)
{
r = await _client.SearchAsync<ReturnWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex)
.Index(GlobalConst.ReturnWarningDetailsIndex)
.Query(q => q.Bool(b => b
.Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status)))
.Should(musts)
@ -463,7 +454,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
{
musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status)));
r = await _client.SearchAsync<ReturnWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex)
.Index(GlobalConst.ReturnWarningDetailsIndex)
.Query(q => q.Bool(b => b
.Must(musts)
))
@ -473,9 +464,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return [];
throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -485,16 +473,13 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
public async Task<List<ReturnMessageRecordIndex>> GetReturnMessageRecordListByIdAsync(string esId)
{
var r = await _client.SearchAsync<ReturnMessageRecordIndex>(x => x
.UseMonthlyIndex(GlobalConst.ReturnMessageRecordIndex)
.Index(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}。");
}
@ -504,13 +489,13 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
public async Task ReturnWarningDetailsRetryIncrementAsync(HashSet<string> esIds)
{
var r = await _client.UpdateByQueryAsync<ReturnWarningDetailsIndex>(u =>
u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex)
u.Index(GlobalConst.ReturnWarningDetailsIndex)
.Query(q => q.Ids(i => i.Values(esIds)))
.Script(s => s
.Source("ctx._source.retryCount = ctx._source.retryCount+1;")
)
);
if (!r.IsValid && !r.IsIndexUnavailable())
if (!r.IsValid)
{
throw new InvalidOperationException($"回传预警明细更新失败，原因：{r.DebugInformation}。");
}
@ -676,7 +661,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<ReturnWarningDetailsIndex>(u =>
u.UseMonthlyIndex(GlobalConst.ReturnWarningDetailsIndex)
u.Index(GlobalConst.ReturnWarningDetailsIndex)
.Query(q => q.Bool(b => b.Must(musts)))
.Script(s => s
.Source("ctx._source.status = params.status")
@ -684,7 +669,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
)
);
if (!r.IsValid && !r.IsIndexUnavailable())
if (!r.IsValid)
{
throw new InvalidOperationException(
$"回传预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。");
@ -781,7 +766,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
.Params(p => p.Add("value", input.MissingTrackStatusId.ToString())))));
var r = await _client.SearchAsync<CompleteWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex)
.Index(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)
@ -791,9 +776,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return new PagedResult<CompleteWarningDetailsIndex>(0, []);
throw new InvalidOperationException($"回传预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -848,7 +830,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (isShouldQuery)
{
r = await _client.SearchAsync<CompleteWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex)
.Index(GlobalConst.CompleteWarningDetailsIndex)
.Query(q => q.Bool(b => b
.Must(mu => mu.Terms(t => t.Field(f => f.Status).Terms(input.Status)))
.Should(musts)
@ -860,7 +842,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
{
musts.Add(c => c.Terms(t => t.Field(f => f.Status).Terms(input.Status)));
r = await _client.SearchAsync<CompleteWarningDetailsIndex>(x => x
.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex)
.Index(GlobalConst.CompleteWarningDetailsIndex)
.Query(q => q.Bool(b => b
.Must(musts)
))
@ -870,9 +852,6 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
if (!r.IsValid)
{
if (r.IsIndexUnavailable())
return [];
throw new InvalidOperationException($"完整预警明细查询失败，原因：{r.DebugInformation}。");
}
@ -893,7 +872,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<CompleteWarningDetailsIndex>(u =>
u.UseMonthlyIndex(GlobalConst.CompleteWarningDetailsIndex)
u.Index(GlobalConst.CompleteWarningDetailsIndex)
.Query(q => q.Bool(b => b.Must(musts)))
.Script(s => s
.Source(@"
@ -907,7 +886,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
)
);
if (!r.IsValid && !r.IsIndexUnavailable())
if (!r.IsValid)
{
throw new InvalidOperationException(
$"完整预警明细更新失败，input：{input.ToJson()}，原因：{r.DebugInformation}。");