using MAX.Core.DependencyInjection; using MAX.Framework.Commons; using MAX.Framework.Dtos; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using OpenSearch.Client; using OpenSearch.Net; using Store.Modules.CloudProduct.Dtos; using Store.Modules.CloudProduct.Enums; namespace Store.Modules.CloudProduct.Services { /// /// 用户收藏商品索引服务:负责收藏索引(Search:FavoritesIndexName)的 /// 创建、写入、删除、重建与检索,索引文档由 user_goods_favorites 关联 goods 补全而来。 /// public interface IUserFavoritesIndexService : ISingletonDependency { /// 检查索引是否存在 Task IndexExistsAsync(); /// 检查索引映射(分析器)是否符合当前服务要求 Task IsIndexMappingHealthyAsync(); /// 创建索引(含 IK 分词映射;已存在且映射正常时跳过) Task CreateIndexAsync(); /// 删除并重建索引(映射统一,重建期间该索引不可查) Task ClearIndexAsync(); /// 批量写入收藏文档(存在则覆盖) Task IndexDocumentsAsync(IEnumerable documents); /// 按文档Id批量删除收藏文档 Task DeleteDocumentsAsync(IEnumerable documentIds); /// 统计索引文档数 Task CountDocumentsAsync(); /// 按用户检索收藏商品(标题/销量/价格/类目/分组) Task<(long Total, List Documents)> SearchAsync(UserFavoritesSearchRequest request, UserFavoritesSortBy sortBy); } internal class UserFavoritesIndexService : IUserFavoritesIndexService { private readonly string _nodeUrl; private readonly string _indexName; private readonly IOpenSearchClient _client = null!; private readonly ILogger _logger; // 查询类操作(search/count/exists)统一超时 private static readonly TimeSpan QueryTimeout = TimeSpan.FromSeconds(5); // 批量写入/建索引/删除索引等管理类操作超时(数据量大时需要更长) private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30); public UserFavoritesIndexService(IConfiguration configuration, ILogger logger) { var config = configuration.GetSection("Search"); _nodeUrl = config.GetValue("Url") ?? ""; _indexName = config.GetValue("FavoritesIndexName") ?? ""; _logger = logger; _client = CreateClient(); } private IOpenSearchClient CreateClient() { var pool = new SingleNodeConnectionPool(new Uri(_nodeUrl)); var settings = new ConnectionSettings(pool) .DefaultIndex(_indexName) .DefaultMappingFor(m => m.IndexName(_indexName)) .RequestTimeout(QueryTimeout) .MaxRetryTimeout(QueryTimeout) .PrettyJson() .DisableDirectStreaming(); return new OpenSearchClient(settings); } /// /// ============ 1. 索引是否存在 ============ /// public async Task IndexExistsAsync() { var response = await _client.Indices.ExistsAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(QueryTimeout))); return response.Exists; } /// /// ============ 2. 索引映射是否健康 ============ /// public async Task IsIndexMappingHealthyAsync() { var getResponse = await _client.Indices.GetAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(QueryTimeout))); if (!getResponse.IsValid) { throw new AppException("读取收藏索引映射失败: " + getResponse.DebugInformation); } if (!getResponse.Indices.TryGetValue(_indexName, out var indexState)) { return false; } var mapping = indexState.Mappings; if (mapping?.Properties == null || !mapping.Properties.TryGetValue(new PropertyName("title"), out var property) || property is not ITextProperty textProperty) { return false; } var indexAnalyzer = textProperty.Analyzer; var searchAnalyzer = textProperty.SearchAnalyzer; var usesIk = indexAnalyzer == "ik_max_word" && searchAnalyzer == "ik_smart"; var usesStandard = indexAnalyzer == "standard" && searchAnalyzer == "standard"; return usesIk || usesStandard; } /// /// ============ 3. 创建索引 ============ /// public async Task CreateIndexAsync() { var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(AdminTimeout))); if (existsResponse.Exists) { if (await IsIndexMappingHealthyAsync()) { return true; } _logger.LogWarning("收藏索引 [{Index}] 的映射不完整,正在重建...", _indexName); var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(AdminTimeout))); if (!deleteResponse.IsValid) { throw new AppException("删除损坏的收藏索引失败: " + deleteResponse.DebugInformation); } } // 优先尝试使用 IK 中文分词(需集群安装 analysis-ik 插件) var createResponse = await TryCreateIndexAsync(useIk: true); if (!createResponse.IsValid) { _logger.LogWarning("IK 分词插件不可用,收藏索引已回退到 standard 分析器,中文分词将失效。"); createResponse = await TryCreateIndexAsync(useIk: false); } if (!createResponse.IsValid) { throw new AppException("收藏索引创建失败: " + createResponse.DebugInformation); } _logger.LogInformation("收藏索引 [{Index}] 创建成功。", _indexName); return true; } private async Task TryCreateIndexAsync(bool useIk) { var indexAnalyzer = useIk ? "ik_max_word" : "standard"; var searchAnalyzer = useIk ? "ik_smart" : "standard"; return await _client.Indices.CreateAsync(_indexName, c => c .Settings(s => s .NumberOfShards(1) .NumberOfReplicas(0) ) .Map(m => m .Properties(p => p .Keyword(k => k.Name(x => x.Id)) .Number(n => n.Name(x => x.FavId).Type(NumberType.Long)) .Keyword(k => k.Name(x => x.UserName)) .Number(n => n.Name(x => x.GroupId).Type(NumberType.Integer)) .Keyword(k => k.Name(x => x.GroupName)) .Number(n => n.Name(x => x.GoodsId).Type(NumberType.Long)) .Number(n => n.Name(x => x.Platform).Type(NumberType.Byte)) .Text(t => t .Name(x => x.Title) .Analyzer(indexAnalyzer) .SearchAnalyzer(searchAnalyzer) ) .Text(t => t.Name(x => x.ThumbUrl)) .Number(n => n.Name(x => x.CatId1).Type(NumberType.Integer)) .Number(n => n.Name(x => x.CatId2).Type(NumberType.Integer)) .Number(n => n.Name(x => x.CatId3).Type(NumberType.Integer)) .Number(n => n.Name(x => x.CatId4).Type(NumberType.Integer)) .Number(n => n.Name(x => x.MinPrice).Type(NumberType.Long)) .Number(n => n.Name(x => x.SoldQuantity).Type(NumberType.Integer)) .Date(d => d.Name(x => x.CreatedAt)) ) ) .RequestConfiguration(r => r.RequestTimeout(AdminTimeout)) ); } /// /// ============ 4. 删除并重建索引 ============ /// public async Task ClearIndexAsync() { var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(QueryTimeout))); if (!existsResponse.Exists) { return await CreateIndexAsync(); } var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(AdminTimeout))); if (!deleteResponse.IsValid) { throw new AppException($"删除收藏索引 {_indexName} 失败: " + deleteResponse.DebugInformation); } _logger.LogInformation("收藏索引 [{Index}] 已删除,准备重建。", _indexName); return await CreateIndexAsync(); } private async Task RefreshIndexAsync() { var response = await _client.Indices.RefreshAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(AdminTimeout))); if (!response.IsValid) { throw new AppException("刷新收藏索引失败: " + response.DebugInformation); } } /// /// ============ 5. 批量写入收藏文档(存在则覆盖) ============ /// public async Task IndexDocumentsAsync(IEnumerable documents) { if (documents == null) throw new ArgumentNullException(nameof(documents)); var list = documents.Where(x => x != null && string.IsNullOrWhiteSpace(x.Id) == false).ToList(); if (list.Count == 0) return; await CreateIndexAsync(); var bulkResponse = await _client.BulkAsync(b => b .Index(_indexName) .IndexMany(list, (descriptor, doc) => descriptor.Id(doc.Id)) .RequestConfiguration(r => r.RequestTimeout(AdminTimeout)) ); if (!bulkResponse.IsValid) { throw new AppException("批量写入收藏索引失败: " + bulkResponse.DebugInformation); } await RefreshIndexAsync(); } /// /// ============ 6. 按文档Id批量删除收藏文档 ============ /// public async Task DeleteDocumentsAsync(IEnumerable documentIds) { var list = documentIds?.Where(x => string.IsNullOrWhiteSpace(x) == false).Distinct().ToList() ?? new List(); if (list.Count == 0) return; if (await IndexExistsAsync() == false) { return; } var response = await _client.BulkAsync(b => { foreach (var id in list) { b.Delete(d => d.Index(_indexName).Id(id)); } b.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)); return b; }); if (!response.IsValid) { throw new AppException("删除收藏索引文档失败: " + response.DebugInformation); } await RefreshIndexAsync(); } /// /// ============ 7. 统计文档数 ============ /// public async Task CountDocumentsAsync() { var countResponse = await _client.CountAsync(c => c .Index(_indexName) .RequestConfiguration(r => r.RequestTimeout(QueryTimeout)) ); if (!countResponse.IsValid) { throw new AppException("统计收藏索引失败: " + countResponse.DebugInformation); } return countResponse.Count; } /// /// ============ 8. 检索用户收藏商品(标题/价格/销量/类目/分组 + 排序 + 分页) ============ /// public async Task<(long Total, List Documents)> SearchAsync(UserFavoritesSearchRequest request, UserFavoritesSortBy sortBy) { if (request == null) throw new ArgumentNullException(nameof(request)); if (string.IsNullOrWhiteSpace(request.UserName)) { throw new AppException("用户名不能为空", ResultCode.ValidationError); } var keyword = request.Title; int? groupId = request.GroupId; var groupName = request.GroupName; int pageIndex = request.PageIndex; int pageSize = request.PageSize; if (pageIndex < 1) pageIndex = 1; if (pageSize < 1) pageSize = 20; if (pageSize > 2000) pageSize = 2000; // 避免深分页/单页过大拖垮性能 var from = (pageIndex - 1) * pageSize; ISearchResponse searchResponse; try { searchResponse = await _client.SearchAsync(s => s .Index(_indexName) .Query(q => q .Bool(b => { var filters = new List, QueryContainer>>(); // 只检索当前用户自己的收藏 —— 放进 filters,避免被后面的 Filter 覆盖 filters.Add(f => f.Term(t => t.Field(x => x.UserName).Value(request.UserName))); if (groupId.HasValue) { filters.Add(f => f.Term(t => t.Field(x => x.GroupId).Value(groupId.Value))); } else if (string.IsNullOrWhiteSpace(groupName) == false) { filters.Add(f => f.Term(t => t.Field(x => x.GroupName).Value(groupName))); } if (request.Platform.HasValue && request.Platform.Value > 0) { filters.Add(f => f.Term(t => t.Field(x => x.Platform).Value(request.Platform.Value))); } if (request.CatId1.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId1).Value(request.CatId1.Value))); if (request.CatId2.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId2).Value(request.CatId2.Value))); if (request.CatId3.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId3).Value(request.CatId3.Value))); if (request.CatId4.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId4).Value(request.CatId4.Value))); if (request.MinPrice.HasValue || request.MaxPrice.HasValue) { filters.Add(f => f.Range(r => { var range = r.Field(x => x.MinPrice); if (request.MinPrice.HasValue) range = range.GreaterThanOrEquals(request.MinPrice.Value); if (request.MaxPrice.HasValue) range = range.LessThanOrEquals(request.MaxPrice.Value); return range; })); } if (request.MinSoldQuantity.HasValue || request.MaxSoldQuantity.HasValue) { filters.Add(f => f.Range(r => { var range = r.Field(x => x.SoldQuantity); if (request.MinSoldQuantity.HasValue) range = range.GreaterThanOrEquals(request.MinSoldQuantity.Value); if (request.MaxSoldQuantity.HasValue) range = range.LessThanOrEquals(request.MaxSoldQuantity.Value); return range; })); } // 统一一次性应用所有 filter,避免覆盖 b.Filter(filters.ToArray()); if (string.IsNullOrWhiteSpace(keyword) == false) { b.Must(m => m.Match(mm => mm.Field(f => f.Title).Query(keyword))); } return b; }) ) .Sort(sort => sortBy switch { UserFavoritesSortBy.CreatedTimeAsc => sort.Ascending(f => f.CreatedAt), UserFavoritesSortBy.PriceAsc => sort.Ascending(f => f.MinPrice), UserFavoritesSortBy.PriceDesc => sort.Descending(f => f.MinPrice), UserFavoritesSortBy.SoldDesc => sort.Descending(f => f.SoldQuantity), _ => sort.Descending(f => f.CreatedAt) }) .From(from) .Size(pageSize) .TrackTotalHits(true) .RequestConfiguration(r => r.RequestTimeout(QueryTimeout)) );} catch (OpenSearchClientException ex) when (ex.Response?.Success == false) { throw new AppException($"收藏检索超时或失败({QueryTimeout.TotalSeconds}秒): " + ex.Message); } if (!searchResponse.IsValid) { throw new AppException("收藏检索失败: " + searchResponse.DebugInformation); } return (searchResponse.Total, searchResponse.Documents.ToList()); } } }