track/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs

using Nest;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Expand.ElasticSearch.Services;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;

namespace SJZY.Track.Repository.Repositories;

internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetailsIndex>, ITrackDetailsRepository,
    ITransientDependency
{
    const int SafelyLimit = GlobalConst.SafelyLimit;
    public TrackDetailsRepository(ConnectionFactory connectionFactory) : base(connectionFactory)
    {
    }

    public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode,
        bool forUpsert = false)
    {
        var r = !forUpsert
            ? await Context.SearchAsync<TrackDetailsIndex>(s => s
                .Source()
                .Size(SafelyLimit)
                .Query(q => q
                    .Bool(b => b
                        .Must(
                            mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
                            mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
                        )
                    )
                )
            )
            : await Context.SearchAsync<TrackDetailsIndex>(s => s
                .Source(s2 => s2.Includes(fd => fd
                    .Field(f => f.IndexName)
                    .Field(f => f.OrderId)
                    .Field(f => f.TrackTime)
                    .Field(f => f.TrackPlace)
                ))
                .Size(SafelyLimit)
                .Query(q => q
                    .Bool(b => b
                        .Must(
                            mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
                            mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
                        )
                    )
                )
            );

        if (!r.IsValid)
        {
            throw new InvalidOperationException(
                $"轨迹查询失败，orderIds：[{orderIds.JoinAsString(", ")}]，trackCode：{trackCode}，原因：{r.DebugInformation}。");
        }

        return r.ToIdList();
    }

    public async Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] orderIds, List<string> trackCodes)
    {
        var excludeTrackCodes = trackCodes.ToList();
        excludeTrackCodes.Add(GlobalConst.EmptyString);
        var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
            .Source(s => s.Includes(fd => fd
                .Field(f => f.TrackEnContent)
                .Field(f => f.TrackContent)
                .Field(f => f.TrackPlace)
                .Field(f => f.TrackTime)
                .Field(f => f.TrackTimeZone)
                .Field(f => f.OrderId)
                .Field(f => f.TrackCode)
            ))
            .Query(q => q
                .Bool(b => b
                    .Must(
                        mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
                    )
                    .MustNot(mn => mn
                        .Terms(t => t
                            .Field(f => f.TrackCode.Suffix("keyword"))
                            .Terms(excludeTrackCodes)
                        )
                    )
                )
            )
            .Size(SafelyLimit)
            .Aggregations(a => a
                .Terms("orders", t => t
                    .Field(f => f.OrderId)
                    .Aggregations(aa => aa
                        .TopHits("latest_track", th => th
                            .Size(1)
                            .Sort(so => so
                                .Field(f => f.ZeroTrackTime, SortOrder.Descending)
                            )
                        )
                    )
                )
            )
        );

        if (!r.IsValid)
        {
            throw new InvalidOperationException(
                $"最新轨迹查询失败，orderIds：[{orderIds.JoinAsString(", ")}]，trackCodes：[{trackCodes.JoinAsString(", ")}]，原因：{r.DebugInformation}。");
        }

        return r.Documents.ToList();
    }

    public async Task<bool> AnyAsync(int orderId, string originTrackDataHash)
    {
        var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
            .Source(s => s.Includes(fd => fd.Field(f => f.TrackCode)))
            .Query(q =>
                q.Term(f => f.OrderId, orderId)
                && q.Term(f => f.OriginTrackDataHash.Suffix("keyword"), originTrackDataHash)
            )
        );

        if (!r.IsValid)
        {
            throw new InvalidOperationException(
                $"轨迹查询失败，orderId：{orderId}，hash：{originTrackDataHash}，原因：{r.DebugInformation}。");
        }

        return r.Hits.Count > 0;
    }


    public async Task UpdateAsync(TrackDetailsIndex track)
    {
        var r = await Context.UpdateAsync<TrackDetailsIndex, object>(track.Id, u => u
            .Index(track.IndexName)
            .Doc(new
            {
                track.TrackPlace,
                track.TrackTime,
                track.TrackTimeZone,
                track.UpdateUser,
                UpdateTime = DateTime.Now,
            })
        );

        if (!r.IsValid)
        {
            throw new InvalidOperationException($"轨迹更新失败，track：{track.ToJson()}，原因：{r.DebugInformation}。");
        }
    }

    public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
    {
        if (!tracks.Any()) return;
        // 按天写入es
        var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
        tracks.ForEach(x => x.IndexName = indexName);

        var r = await Context.IndexManyAsync(tracks, indexName);

        if (!r.IsValid)
        {
            throw new InvalidOperationException($"{tracks.Count}条轨迹新增失败，原因：{r.DebugInformation}。");
        }
    }

    public async Task AddAsync(TrackDetailsIndex track)
    {
        // 按天写入es
        var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
        track.IndexName = indexName;
        var r = await Context.IndexAsync(track, s => s.Index(indexName));

        if (!r.IsValid)
        {
            throw new InvalidOperationException($"轨迹新增失败，track：{track.ToJson()}，原因：{r.DebugInformation}。");
        }
    }

    public async Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos)
    {
        if (orderNos.Count < 1) return new List<TrackDetailsIndex>();

        var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
            .Source(s => s.Includes(fd => fd
                .Field(f => f.OrderNo)
                .Field(f => f.TrackStatus)
                .Field(f => f.TrackCode)
                .Field(f => f.TrackContent)
                .Field(f => f.OriginTrackContent)
                .Field(f => f.TrackPlace)
                .Field(f => f.TrackTime)
                .Field(f => f.TrackTimeZone)
            ))
            .Query(q =>
                q.Bool(b => b
                    .Must(mu => mu
                            .Terms(t => t
                                .Field(f => f.OrderNo.Suffix("keyword"))
                                .Terms(orderNos)
                            ),
                        mu => mu
                            .Term(f => f.OrderType, orderType)
                    )
                    // tips：对外不暴露服务商轨迹
                    .MustNot(mn => mn
                        .Term(t => t.TrackCode.Suffix("keyword"), GlobalConst.EmptyString)
                    )
                )
            )
            .Size(SafelyLimit)
        );

        if (!r.IsValid)
        {
            throw new InvalidOperationException(
                $"轨迹查询失败，orderType：{orderType}，orderNos：{orderNos}，原因：{r.DebugInformation}。");
        }

        return r.Documents.ToList();
    }

    public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)
    {
        var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
            .Query(q =>
                q.Term(t => t.OrderId, orderId)
            )
            .Size(SafelyLimit)
            .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending)));

        if (!r.IsValid)
        {
            throw new InvalidOperationException($"轨迹查询失败，orderId：{orderId}，原因：{r.DebugInformation}。");
        }

        return r.ToIdList();
    }

    public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
    {
        var search = await Context.SearchAsync<TrackDetailsIndex>(x => x
            .Source(s => s.Includes(fd => fd.Field(f => f.IndexName)))
            .Size(SafelyLimit)
            .Query(q =>
                q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
                && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))
            )
        );

        if (!search.IsValid)
        {
            throw new InvalidOperationException(
                $"轨迹查询失败，orderIds：[{orderIds.JoinAsString(", ")}]，trackCodes：[{trackCodes.JoinAsString(", ")}]，原因：{search.DebugInformation}。");
        }

        var tracks = search.ToIdList();

        foreach (var track in tracks)
        {
            await DeleteTrackByIdAsync(track.Id, track.IndexName);
        }
    }

    public async Task DeleteTrackByIdAsync(string id, string indexName)
    {
        var r = await Context.DeleteAsync<TrackDetailsIndex>(id, u => u
            .Index(indexName)
        );

        if (!r.IsValid)
        {
            throw new InvalidOperationException($"轨迹删除失败，id：{indexName}，原因：{r.DebugInformation}。");
        }
    }
}