diff --git a/Modules/CloudProduct/CloudProductModule.cs b/Modules/CloudProduct/CloudProductModule.cs index 24ead13..28b3728 100644 --- a/Modules/CloudProduct/CloudProductModule.cs +++ b/Modules/CloudProduct/CloudProductModule.cs @@ -64,6 +64,9 @@ namespace Store.Modules.CloudProduct .UseStorage(new MySqlStorage(connectionStringBuilder.ConnectionString, new MySqlStorageOptions() { TablesPrefix = "Hangfire_" }))); services.AddHangfireServer(); services.AddTransient(); + services.AddTransient(); + services.AddAuthorization(options => options.AddPolicy("CloudGoodsViolationWrite", + policy => policy.RequireAuthenticatedUser().RequireClaim("permission", "cloudgoods:violation:write"))); } @@ -96,6 +99,8 @@ namespace Store.Modules.CloudProduct "goods-index-pending", job => job.ProcessPendingAsync(), cron); + app.ApplicationServices.GetRequiredService().AddOrUpdate( + "goods-violation-delete-pending", job => job.ProcessPendingAsync(), "* * * * *"); } catch (Exception ex) { diff --git a/Modules/CloudProduct/Controllers/GoodsController.cs b/Modules/CloudProduct/Controllers/GoodsController.cs index db57ec9..38bf367 100644 --- a/Modules/CloudProduct/Controllers/GoodsController.cs +++ b/Modules/CloudProduct/Controllers/GoodsController.cs @@ -2,6 +2,7 @@ using Asp.Versioning; using MAX.Framework.Commons; using MAX.Framework.Dtos; using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Mvc; using Store.Framework.Base; using Store.Modules.CloudProduct.Dtos; @@ -12,6 +13,8 @@ using System.Text.Json.Nodes; namespace Store.Modules.CloudProduct.Controllers { + /// 云商品管理：商品查询、导入、违规上报及搜索索引维护。 + [Tags("云商品")] [ApiVersionNeutral] [Route("api/CloudGoods/[action]")] public class GoodsController : AppControllerBase @@ -20,31 +23,62 @@ namespace Store.Modules.CloudProduct.Controllers private readonly IOpenSearchService _searchService; private readonly IRawDataStorage _rawDataStorage; private readonly GoodsIndexJob _goodsIndexJob; + private readonly GoodsViolationService _violations; - public GoodsController(IGoodsService goodsService, IOpenSearchService searchService, IRawDataStorage rawDataStorage, GoodsIndexJob goodsIndexJob) + public GoodsController(IGoodsService goodsService, IOpenSearchService searchService, IRawDataStorage rawDataStorage, GoodsIndexJob goodsIndexJob, GoodsViolationService violations) { + _violations = violations; _goodsService = goodsService; _searchService = searchService; _rawDataStorage = rawDataStorage; _goodsIndexJob = goodsIndexJob; } + /// 获取全部违规原因字典，包含未启用违规判定的原因，按ID升序返回。 [HttpGet] + [EndpointSummary("获取违规原因字典")] + public Task> GetViolationReasons() + => _violations.GetReasonsAsync(); + + /// 批量上报违规商品；重复上报合并原因。 + [HttpPost] + [EndpointSummary("批量上报违规商品")] + public Task> ReportViolations([FromBody] ReportGoodsViolationItem[] items) + => _violations.ReportAsync(items); + + /// 批量解除商品违规，恢复正常商品索引；失败后由后台索引任务重试。 + [HttpPost] + [EndpointSummary("解除商品违规并恢复索引")] + public Task RemoveViolations([FromBody] long[] ids) + => _violations.RemoveAsync(ids); + + /// 按商品编号批量查询是否违规。 + [HttpPost] + [EndpointSummary("批量查询商品违规状态")] + public Task> GetViolationStatus([FromBody] long[] ids) + => _violations.GetStatusAsync(ids); + + /// 获取服务器当前日期（长日期格式）。 + [HttpGet] + [EndpointSummary("获取服务器当前日期")] public async Task Time() { return DateTime.Now.ToLongDateString(); } + /// 手动执行待处理商品的索引任务。 [HttpGet] + [EndpointSummary("处理待建立索引的商品")] public async Task Job() { await _goodsIndexJob.ProcessPendingAsync(); } /// - /// 搜索 + /// 分页搜索商品，可按标题、价格、类目、销量和平台筛选。 /// - /// - /// + /// 商品筛选条件及分页参数。 + /// 符合条件的商品列表及总数。 [HttpPost] + [EndpointSummary("分页搜索商品")] public async Task> SearchGoods(GoodsSearchRequest request) { // 1. 参数基础校验 @@ -145,6 +179,7 @@ namespace Store.Modules.CloudProduct.Controllers /// [HttpGet] [ActionName("SearchGoodsToIds")] + [EndpointSummary("按关键词搜索商品摘要")] public async Task> SearchGoodsSummary(string keyword, int pageindex = 1, int pagesize = 20) { if (string.IsNullOrWhiteSpace(keyword)) @@ -172,6 +207,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 导入测试数据：读取 ImportConverters/data.txt 并走完整解析 + 入库 + 分词流程。 /// [HttpGet] + [EndpointSummary("导入测试商品数据")] public async Task ImportTestData() { var filePath = GetDataFilePath(); @@ -198,6 +234,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 导入接口：接收客户端批量上传的 JSON 文件，解析后入库并写入分词索引。 /// [HttpPost] + [EndpointSummary("批量导入商品 JSON 文件")] public async Task ImportFiles([FromForm] List files) { if (files == null || files.Count == 0) @@ -247,6 +284,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 查询 OpenSearch 状态。 /// [HttpGet] + [EndpointSummary("查询 OpenSearch 服务状态")] public async Task GetOpenSearchStatus() { return await _searchService.GetStatusAsync(); @@ -256,6 +294,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 检查搜索索引是否存在。 /// [HttpGet] + [EndpointSummary("检查商品搜索索引是否存在")] public async Task IndexExists() { return await _searchService.IndexExistsAsync(); @@ -265,6 +304,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 检查索引映射（分析器）是否符合当前服务要求。 /// [HttpGet] + [EndpointSummary("检查商品索引映射是否正常")] public async Task IndexMappingHealthy() { return await _searchService.IsIndexMappingHealthyAsync(); @@ -274,6 +314,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 创建搜索索引（含 IK 分词映射，已存在且映射正常时跳过）。 /// [HttpPost] + [EndpointSummary("创建商品搜索索引")] public async Task CreateIndex() { return await _searchService.CreateIndexAsync(); @@ -283,6 +324,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 删除搜索索引。 /// [HttpPost] + [EndpointSummary("删除商品搜索索引")] public async Task DeleteIndex() { await _searchService.DeleteIndexAsync(); @@ -292,6 +334,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 手动刷新搜索索引，使最新写入立即可搜索。 /// [HttpPost] + [EndpointSummary("刷新商品搜索索引")] public async Task RefreshIndex() { await _searchService.RefreshIndexAsync(); @@ -301,6 +344,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 手动新增或覆盖一条搜索文档。 /// [HttpPost] + [EndpointSummary("新增或覆盖商品搜索文档")] public async Task AddSearchDocument([FromBody] GoodsSearchDocument document) { if (document == null || document.Id <= 0) @@ -315,6 +359,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 读取指定商品详情：解压缩原始数据并返回 JSON。 /// [HttpPost] + [EndpointSummary("查询商品详情")] public async Task GetGoodsDetail([FromQuery] long goodsId) { var raw = await GetRawDataOrThrowAsync(goodsId); @@ -326,6 +371,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 读取指定商品的原始数据（解压缩后原样返回）。 /// [HttpPost] + [EndpointSummary("读取商品原始数据")] public async Task GetGoodsRawData([FromQuery] long goodsId) { return await GetRawDataOrThrowAsync(goodsId); @@ -336,6 +382,7 @@ namespace Store.Modules.CloudProduct.Controllers /// ids 为空时重建全部商品。 /// [HttpPost] + [EndpointSummary("重建商品搜索文档")] public async Task RebuildIndex([FromBody] long[]? ids) { return await _goodsService.RebuildSearchIndexAsync(ids); @@ -346,6 +393,7 @@ namespace Store.Modules.CloudProduct.Controllers /// 软删除将 Status 置 1，硬删除物理删除记录；两种方式都会删除 OpenSearch 中对应文档。 /// [HttpPost] + [EndpointSummary("批量删除商品")] public async Task DeleteGoods([FromBody] long[]? ids) { if (ids == null || ids.Length == 0) @@ -361,6 +409,7 @@ namespace Store.Modules.CloudProduct.Controllers /// [HttpGet] [ActionName("ClearSearch")] + [EndpointSummary("清空并重建商品搜索索引")] public async Task ClearSearchIndex() { return await _searchService.ClearIndexAsync(); @@ -371,6 +420,7 @@ namespace Store.Modules.CloudProduct.Controllers /// [HttpGet] [ActionName("GetSearchCount")] + [EndpointSummary("统计商品搜索文档数量")] public async Task GetSearchDocumentCount() { return await _searchService.CountDocumentsAsync(); diff --git a/Modules/CloudProduct/Dtos/GoodsViolationDtos.cs b/Modules/CloudProduct/Dtos/GoodsViolationDtos.cs new file mode 100644 index 0000000..371ea69 --- /dev/null +++ b/Modules/CloudProduct/Dtos/GoodsViolationDtos.cs @@ -0,0 +1,36 @@ +namespace Store.Modules.CloudProduct.Dtos; + +public class ViolationReasonDto +{ + public string Name { get; set; } = string.Empty; + public bool IsViolation { get; set; } +} + +public class ReportGoodsViolationItem +{ + public long GoodsId { get; set; } + /// 原因名称，例如“出售假冒商品”；新名称默认仅记录字典，不判定违规。 + public string Reasons { get; set; } = string.Empty; +} + +public class ReportGoodsViolationsResult +{ + public long GoodsId { get; set; } + /// 存在违规记录且已完成索引下架时为 true，否则为 false。 + public bool IsOffShelf { get; set; } +} + +public class GoodsViolationStatus +{ + public long GoodsId { get; set; } + public bool IsViolation { get; set; } +} + +public class RemoveGoodsViolationsResult +{ + public int RemovedCount { get; set; } + public long[] RestoredGoodsIds { get; set; } = []; + public long[] PendingRestoreGoodsIds { get; set; } = []; + /// 商品不存在、已删除或恢复期间再次被上报违规的编号。 + public long[] SkippedGoodsIds { get; set; } = []; +} diff --git a/Modules/CloudProduct/Entities/GoodsViolationEntity.cs b/Modules/CloudProduct/Entities/GoodsViolationEntity.cs new file mode 100644 index 0000000..6626067 --- /dev/null +++ b/Modules/CloudProduct/Entities/GoodsViolationEntity.cs @@ -0,0 +1,22 @@ +using FreeSql.DataAnnotations; + +namespace Store.Modules.CloudProduct.Entities; + +[Table(Name = "goods_violation")] +public class GoodsViolationEntity +{ + [Column(Name = "goods_id", IsPrimary = true)] + public long GoodsId { get; set; } + + [Column(Name = "reason_id", IsPrimary = true, IsNullable = false)] + public uint ReasonId { get; set; } + + [Column(Name = "is_index_deleted")] + public bool IsIndexDeleted { get; set; } + + [Column(Name = "created_time")] + public DateTime CreatedTime { get; set; } + + [Column(Name = "updated_time")] + public DateTime UpdatedTime { get; set; } +} diff --git a/Modules/CloudProduct/Entities/ViolationReasonEntity.cs b/Modules/CloudProduct/Entities/ViolationReasonEntity.cs new file mode 100644 index 0000000..3909b97 --- /dev/null +++ b/Modules/CloudProduct/Entities/ViolationReasonEntity.cs @@ -0,0 +1,16 @@ +using FreeSql.DataAnnotations; + +namespace Store.Modules.CloudProduct.Entities; + +[Table(Name = "violation_reason")] +public class ViolationReasonEntity +{ + [Column(Name = "id", IsPrimary = true, IsIdentity = true)] + public uint Id { get; set; } + + [Column(Name = "name", StringLength = 100, IsNullable = false)] + public string Name { get; set; } = string.Empty; + + [Column(Name = "is_violation", IsNullable = false)] + public bool IsViolation { get; set; } +} diff --git a/Modules/CloudProduct/Repositorys/GoodsViolationRepository.cs b/Modules/CloudProduct/Repositorys/GoodsViolationRepository.cs new file mode 100644 index 0000000..02f9994 --- /dev/null +++ b/Modules/CloudProduct/Repositorys/GoodsViolationRepository.cs @@ -0,0 +1,118 @@ +using FreeSql; +using MAX.Core.DependencyInjection; +using Microsoft.Extensions.DependencyInjection; +using Store.Modules.CloudProduct.Entities; +using Store.Modules.CloudProduct.Dtos; + +namespace Store.Modules.CloudProduct.Repositorys; + +public class GoodsViolationRepository( + [FromKeyedServices("CloudProductConnection")] IFreeSql fsql) + : BaseRepository(fsql), IScopedDependency +{ + public Task> GetReasonsAsync() => Orm.Select() + .OrderBy(x => x.Id).ToListAsync(x => new ViolationReasonDto + { + Name = x.Name, + IsViolation = x.IsViolation + }); + + public Task> GetViolationIdsAsync(long[] ids) => Select + .Where(x => ids.Contains(x.GoodsId)).Distinct().ToListAsync(x => x.GoodsId); + + public async Task> GetOffShelfGoodsIdsAsync(long[] ids) + { + var records = await Select.Where(x => ids.Contains(x.GoodsId)).ToListAsync(); + return records.GroupBy(x => x.GoodsId).Where(x => x.All(row => row.IsIndexDeleted)) + .Select(x => x.Key).ToList(); + } + + // 解除违规和设置待同步必须一起提交，进程中断后后台索引任务仍可恢复。 + public async Task RemoveAndScheduleRestoreAsync(long[] ids) + { + using var unit = Orm.CreateUnitOfWork(); + var violations = unit.GetRepository(); + var goods = unit.GetRepository(); + var removedIds = await violations.Select.Where(x => ids.Contains(x.GoodsId)).Distinct().ToListAsync(x => x.GoodsId); + await violations.DeleteAsync(x => ids.Contains(x.GoodsId)); + await goods.UpdateDiy.Set(x => x.IsSegmented, (byte)0) + .Where(x => ids.Contains(x.GoodsId) && x.Status == 0).ExecuteAffrowsAsync(); + unit.Commit(); + return removedIds.Count; + } + + public Task> GetRestorableGoodsAsync(long[] ids) => Orm.Select() + .Where(x => ids.Contains(x.GoodsId) && x.Status == 0).ToListAsync(); + + public Task MarkRestoredAsync(long[] ids) => Orm.Update() + .Set(x => x.IsSegmented, (byte)1) + .Where(x => ids.Contains(x.GoodsId) && x.Status == 0).ExecuteAffrowsAsync(); + + // 调用方持有 GoodsIndexLock，事务内合并原因，避免并发覆盖。 + public async Task MergeAsync(Dictionary items) + { + using var unit = Orm.CreateUnitOfWork(); + var repository = unit.GetRepository(); + var reasonRepository = unit.GetRepository(); + var reasonIds = new Dictionary(StringComparer.Ordinal); + foreach (var name in items.Values.SelectMany(x => x).Distinct(StringComparer.Ordinal)) + { + // 让数据库按字典表的排序规则匹配名称，与唯一索引保持一致。 + var reason = await reasonRepository.Select.Where(x => x.Name == name).FirstAsync(); + // 新类型只收录到字典，绝不因首次上报自动启用商品封禁。 + reason ??= await reasonRepository.InsertAsync(new ViolationReasonEntity { Name = name, IsViolation = false }); + // 数据库排序规则可能忽略部分字符差异；判定时仍要求名称完整相等。 + if (reason.IsViolation && string.Equals(reason.Name, name, StringComparison.Ordinal)) + reasonIds.Add(name, reason.Id); + } + var ids = items.Keys.ToArray(); + var existing = (await repository.Select.Where(x => ids.Contains(x.GoodsId)) + .ToListAsync()).ToLookup(x => x.GoodsId); + var now = DateTime.UtcNow; + var recordedIds = new List(); + foreach (var (goodsId, reasons) in items) + { + var accepted = reasons.Where(reasonIds.ContainsKey).Select(x => reasonIds[x]).Distinct().ToArray(); + // 忽略原因仅记录字典，不新增或改变该商品已有的违规状态。 + if (accepted.Length == 0) continue; + var rows = existing[goodsId].ToArray(); + var previous = rows.Select(x => x.ReasonId).ToHashSet(); + var added = accepted.Where(x => !previous.Contains(x)).ToArray(); + if (added.Length == 0 && rows.All(x => x.IsIndexDeleted)) continue; + + // 下架是商品级操作：新增原因或存在失败记录时，整件商品重新标记待下架。 + foreach (var row in rows) + { + row.IsIndexDeleted = false; + row.UpdatedTime = now; + await repository.UpdateAsync(row); + } + foreach (var reasonId in added) + { + await repository.InsertAsync(new GoodsViolationEntity + { + GoodsId = goodsId, ReasonId = reasonId, IsIndexDeleted = false, + CreatedTime = now, UpdatedTime = now + }); + } + recordedIds.Add(goodsId); + } + unit.Commit(); + return recordedIds.ToArray(); + } + + public Task MarkDeletedAsync(long[] ids) => Orm.Update() + .Set(x => x.IsIndexDeleted, true) + .Set(x => x.UpdatedTime, DateTime.UtcNow) + .Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync(); + + public Task> GetBatchAsync(long afterId, int size, bool pendingOnly) => Select + .Where(x => x.GoodsId > afterId) + .WhereIf(pendingOnly, x => !x.IsIndexDeleted) + .Distinct().OrderBy(x => x.GoodsId).Limit(size).ToListAsync(x => x.GoodsId); + + public Task> GetByReasonAsync(uint reasonId, long afterId, int size) => Select + .Where(x => x.ReasonId == reasonId) + .Where(x => x.GoodsId > afterId).OrderBy(x => x.GoodsId) + .Limit(size).ToListAsync(x => x.GoodsId); +} diff --git a/Modules/CloudProduct/Services/GoodsIndexLock.cs b/Modules/CloudProduct/Services/GoodsIndexLock.cs new file mode 100644 index 0000000..6f72fe5 --- /dev/null +++ b/Modules/CloudProduct/Services/GoodsIndexLock.cs @@ -0,0 +1,41 @@ +using MAX.Core.DependencyInjection; +using MAX.Framework.Commons; +using Microsoft.Extensions.Configuration; +using MySql.Data.MySqlClient; +using System.Security.Cryptography; +using System.Text; + +namespace Store.Modules.CloudProduct.Services; + +/// 跨实例串行化违规上报和商品索引写入，防止下架后被旧批次写回。 +public class GoodsIndexLock(IConfiguration configuration) : ISingletonDependency +{ + public async Task AcquireAsync() + { + // 禁用连接池：即使释放锁失败，关闭物理连接也会释放会话锁。 + var settings = new MySqlConnectionStringBuilder(configuration["CloudProductConnection"]) + { + Pooling = false + }; + var connection = new MySqlConnection(settings.ConnectionString); + var name = "goods-index:" + Convert.ToHexString(SHA256.HashData( + Encoding.UTF8.GetBytes(settings.Database)))[..40]; + try + { + await connection.OpenAsync(); + using var command = connection.CreateCommand(); + command.CommandText = "SELECT GET_LOCK(@name, 30)"; + command.CommandTimeout = 35; + command.Parameters.AddWithValue("@name", name); + var result = await command.ExecuteScalarAsync(); + if (result == null || result == DBNull.Value || Convert.ToInt32(result) != 1) + throw new AppException("商品索引正在处理，请稍后重试"); + return connection; + } + catch + { + await connection.DisposeAsync(); + throw; + } + } +} diff --git a/Modules/CloudProduct/Services/GoodsService.cs b/Modules/CloudProduct/Services/GoodsService.cs index 895e643..438b44d 100644 --- a/Modules/CloudProduct/Services/GoodsService.cs +++ b/Modules/CloudProduct/Services/GoodsService.cs @@ -23,9 +23,11 @@ namespace Store.Modules.CloudProduct.Services private readonly IOpenSearchService _searchService; private readonly IRawDataStorage _rawDataStorage; private readonly IConfiguration _configuration; + private readonly GoodsViolationService _violations; - public GoodsService(GoodsRepository repository, IOpenSearchService searchService, IRawDataStorage rawDataStorage, IConfiguration configuration) + public GoodsService(GoodsRepository repository, IOpenSearchService searchService, IRawDataStorage rawDataStorage, IConfiguration configuration, GoodsViolationService violations) { + _violations = violations; _repository = repository; _searchService = searchService; _rawDataStorage = rawDataStorage; @@ -146,7 +148,6 @@ namespace Store.Modules.CloudProduct.Services var indexImmediately = IsImmediateIndexing(); var documents = new List(); - var indexedIds = new List(); foreach (var item in items) { @@ -194,7 +195,6 @@ namespace Store.Modules.CloudProduct.Services MinPrice = saved.MinPrice, SoldQuantity = saved.SoldQuantity, }); - indexedIds.Add(saved.GoodsId); } result.SuccessCount++; @@ -210,7 +210,7 @@ namespace Store.Modules.CloudProduct.Services { try { - await _searchService.BulkIndexGoodsAsync(documents); + var indexedIds = await _searchService.BulkIndexGoodsAsync(documents); await MarkIndexedAsync(indexedIds); } catch (Exception ex) @@ -233,8 +233,9 @@ namespace Store.Modules.CloudProduct.Services if (batchSize <= 0) batchSize = 200; - var pending = await _repository.Select + var pending = await _repository.Select.As("g") .Where(x => x.IsSegmented == 0 && x.Status == 0) + .Where("NOT EXISTS (SELECT 1 FROM goods_violation v WHERE v.goods_id = g.goods_id)") .OrderBy(x => x.GoodsId) .Limit(batchSize) .ToListAsync(); @@ -258,10 +259,9 @@ namespace Store.Modules.CloudProduct.Services }) .ToList(); - await _searchService.BulkIndexGoodsAsync(documents); - - await MarkIndexedAsync(pending.Select(x => x.GoodsId)); - return pending.Count; + var indexedIds = await _searchService.BulkIndexGoodsAsync(documents); + await MarkIndexedAsync(indexedIds); + return indexedIds.Length; } /// @@ -308,22 +308,22 @@ namespace Store.Modules.CloudProduct.Services /// public async Task RebuildSearchIndexAsync(IEnumerable? ids = null) { - List goodsList; - - var idList = ids?.ToList(); - if (idList == null || idList.Count == 0) + await _violations.DeleteViolationIndexesAsync(pendingOnly: false); + var idList = ids?.Distinct().ToArray() ?? []; + long lastId = 0, count = 0; + while (true) { - goodsList = await _repository.Select.ToListAsync(); + var goodsList = await _repository.Select.As("g") + .Where(x => x.Status == 0 && x.GoodsId > 0 && x.Id > lastId) + .WhereIf(idList.Length > 0, x => idList.Contains(x.GoodsId)) + .Where("NOT EXISTS (SELECT 1 FROM goods_violation v WHERE v.goods_id = g.goods_id)") + .OrderBy(x => x.Id).Limit(1000).ToListAsync(); + if (goodsList.Count == 0) break; + var indexedIds = await _searchService.RebuildIndexAsync(goodsList); + await MarkIndexedAsync(indexedIds); + count += indexedIds.Length; + lastId = goodsList[^1].Id; } - else - { - goodsList = await _repository.Select - .Where(x => idList.Contains(x.GoodsId)) - .ToListAsync(); - } - - var count = await _searchService.RebuildIndexAsync(goodsList); - await MarkIndexedAsync(goodsList.Where(x => x.GoodsId > 0).Select(x => x.GoodsId)); return count; } diff --git a/Modules/CloudProduct/Services/GoodsViolationIndexJob.cs b/Modules/CloudProduct/Services/GoodsViolationIndexJob.cs new file mode 100644 index 0000000..f7d9959 --- /dev/null +++ b/Modules/CloudProduct/Services/GoodsViolationIndexJob.cs @@ -0,0 +1,13 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace Store.Modules.CloudProduct.Services; + +public class GoodsViolationIndexJob(IServiceScopeFactory scopeFactory) +{ + public async Task ProcessPendingAsync() + { + using var scope = scopeFactory.CreateScope(); + await scope.ServiceProvider.GetRequiredService() + .DeleteViolationIndexesAsync(); + } +} diff --git a/Modules/CloudProduct/Services/GoodsViolationService.cs b/Modules/CloudProduct/Services/GoodsViolationService.cs new file mode 100644 index 0000000..c624813 --- /dev/null +++ b/Modules/CloudProduct/Services/GoodsViolationService.cs @@ -0,0 +1,124 @@ +using MAX.Core.DependencyInjection; +using MAX.Framework.Commons; +using MAX.Framework.Dtos; +using Microsoft.Extensions.Logging; +using Store.Modules.CloudProduct.Dtos; +using Store.Modules.CloudProduct.Repositorys; + +namespace Store.Modules.CloudProduct.Services; + +public class GoodsViolationService( + GoodsViolationRepository repository, + GoodsIndexLock indexLock, + IOpenSearchService search, + ILogger logger) : IScopedDependency +{ + public Task> GetReasonsAsync() => repository.GetReasonsAsync(); + + public static Dictionary ValidateAndGroup(ReportGoodsViolationItem[]? items) + { + if (items == null || items.Length is < 1 or > 1000) + throw new AppException("每次请提供1-1000条违规记录", ResultCode.ValidationError); + for (var i = 0; i < items.Length; i++) + { + var item = items[i]; + if (item == null || item.GoodsId <= 0 || string.IsNullOrWhiteSpace(item.Reasons) + || item.Reasons.Trim().Length > 100) + throw new AppException($"第{i + 1}条记录须提供有效商品编号和一个违规原因，原因去除首尾空白后须为1-100个字符", ResultCode.ValidationError); + } + return items.GroupBy(x => x.GoodsId).ToDictionary(x => x.Key, + x => x.Select(y => y.Reasons.Trim()).Distinct(StringComparer.Ordinal).ToArray()); + } + + public async Task> ReportAsync(ReportGoodsViolationItem[]? items) + { + var grouped = ValidateAndGroup(items); + await using var lease = await indexLock.AcquireAsync(); + var pendingIds = await repository.MergeAsync(grouped); + if (pendingIds.Length > 0) + { + try + { + await search.DeleteDocumentsAsync(pendingIds); + await repository.MarkDeletedAsync(pendingIds); + } + catch (Exception ex) + { + logger.LogError(ex, "违规记录已保存，{Count}个商品等待重试下架", pendingIds.Length); + } + } + // 返回商品当前持久化状态，包括本次原因被忽略但历史已下架的商品。 + var offShelfIds = (await repository.GetOffShelfGoodsIdsAsync(grouped.Keys.ToArray())).ToHashSet(); + return grouped.Keys.Select(id => new ReportGoodsViolationsResult + { + GoodsId = id, + IsOffShelf = offShelfIds.Contains(id) + }).ToList(); + } + + public async Task> GetStatusAsync(long[]? ids) + { + if (ids == null || ids.Length is < 1 or > 1000 || ids.Any(x => x <= 0)) + throw new AppException("请提供1-1000个正数商品编号", ResultCode.ValidationError); + var distinct = ids.Distinct().ToArray(); + var violations = (await repository.GetViolationIdsAsync(distinct)).ToHashSet(); + return distinct.Select(x => new GoodsViolationStatus { GoodsId = x, IsViolation = violations.Contains(x) }).ToList(); + } + + public async Task RemoveAsync(long[]? ids) + { + if (ids == null || ids.Length is < 1 or > 1000 || ids.Any(x => x <= 0)) + throw new AppException("请提供1-1000个正数商品编号", ResultCode.ValidationError); + var distinct = ids.Distinct().ToArray(); + var result = new RemoveGoodsViolationsResult(); + await using (var lease = await indexLock.AcquireAsync()) + result.RemovedCount = await repository.RemoveAndScheduleRestoreAsync(distinct); + + // 索引服务自行持锁并再次检查违规，避免与新的上报交错时重新上架违规商品。 + var goods = await repository.GetRestorableGoodsAsync(distinct); + var candidates = goods.Select(x => x.GoodsId).Distinct().ToArray(); + result.SkippedGoodsIds = distinct.Except(candidates).ToArray(); + if (candidates.Length == 0) return result; + try + { + var restored = await search.RebuildIndexAsync(goods); + if (restored.Length > 0) + { + await using var lease = await indexLock.AcquireAsync(); + var blocked = await repository.GetViolationIdsAsync(restored); + restored = restored.Except(blocked).ToArray(); + if (restored.Length > 0) await repository.MarkRestoredAsync(restored); + } + result.RestoredGoodsIds = restored; + result.SkippedGoodsIds = distinct.Except(restored).ToArray(); + } + catch (Exception ex) + { + logger.LogError(ex, "已解除违规，{Count}个商品等待后台任务恢复索引", candidates.Length); + result.PendingRestoreGoodsIds = candidates; + } + return result; + } + + // 暂不注册 HTTP 路由。 + public Task> GetGoodsIdsByReasonAsync(uint reasonId, long afterGoodsId = 0, int pageSize = 100) + { + if (reasonId == 0 || afterGoodsId < 0 || pageSize is < 1 or > 1000) + throw new AppException("原因ID或分页参数无效", ResultCode.ValidationError); + return repository.GetByReasonAsync(reasonId, afterGoodsId, pageSize); + } + + public async Task DeleteViolationIndexesAsync(bool pendingOnly = true) + { + long afterId = 0; + while (true) + { + await using var lease = await indexLock.AcquireAsync(); + var ids = await repository.GetBatchAsync(afterId, 1000, pendingOnly); + if (ids.Count == 0) return; + await search.DeleteDocumentsAsync(ids); + await repository.MarkDeletedAsync(ids.ToArray()); + afterId = ids[^1]; + } + } +} diff --git a/Modules/CloudProduct/Services/OpenSearchService.cs b/Modules/CloudProduct/Services/OpenSearchService.cs index ead187c..633be5b 100644 --- a/Modules/CloudProduct/Services/OpenSearchService.cs +++ b/Modules/CloudProduct/Services/OpenSearchService.cs @@ -2,6 +2,7 @@ using System.Diagnostics; using MAX.Core.DependencyInjection; using MAX.Framework.Commons; using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using OpenSearch.Client; using OpenSearch.Net; @@ -14,7 +15,7 @@ namespace Store.Modules.CloudProduct.Services public interface IOpenSearchService : ISingletonDependency { Task AddDocumentAsync(GoodsSearchDocument document); - Task BulkIndexGoodsAsync(IEnumerable documents); + Task BulkIndexGoodsAsync(IEnumerable documents); Task ClearIndexAsync(); Task CountDocumentsAsync(); Task CreateIndexAsync(); @@ -24,7 +25,7 @@ namespace Store.Modules.CloudProduct.Services Task IndexExistsAsync(); Task IsIndexMappingHealthyAsync(); Task RefreshIndexAsync(); - Task RebuildIndexAsync(IEnumerable goodsList); + Task RebuildIndexAsync(IEnumerable goodsList); Task<(long Total, List Documents)> SearchByTitleAsync(GoodsSearchRequest request, GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance); } @@ -39,9 +40,14 @@ namespace Store.Modules.CloudProduct.Services // 批量写入/建索引/删除索引等管理类操作超时（数据量大时需要更长，可挪到配置文件） private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30); private readonly ILogger _logger; + private readonly IFreeSql _database; + private readonly GoodsIndexLock _indexLock; - public OpenSearchService(IConfiguration configuration,ILogger logger) + public OpenSearchService(IConfiguration configuration,ILogger logger, + [FromKeyedServices("CloudProductConnection")] IFreeSql database, GoodsIndexLock indexLock) { + _database = database; + _indexLock = indexLock; var config = configuration.GetSection("Search"); _nodeUrl = config.GetValue("Url") ?? ""; _indexName = config.GetValue("IndexName") ?? ""; @@ -204,12 +210,18 @@ namespace Store.Modules.CloudProduct.Services /// /// ============ 3.1 批量写入商品搜索文档（存在则覆盖） ============ /// - public async Task BulkIndexGoodsAsync(IEnumerable documents) + public async Task BulkIndexGoodsAsync(IEnumerable documents) { if (documents == null) throw new ArgumentNullException(nameof(documents)); var list = documents.ToList(); - if (list.Count == 0) return; + if (list.Count == 0) return []; + await using var lease = await _indexLock.AcquireAsync(); + var ids = list.Select(x => x.Id).Distinct().ToArray(); + var blocked = (await _database.Select() + .Where(x => ids.Contains(x.GoodsId)).ToListAsync(x => x.GoodsId)).ToHashSet(); + list = list.Where(x => !blocked.Contains(x.Id)).ToList(); + if (list.Count == 0) return []; // 确保索引已按 IK 分词映射创建，否则中文分词无效 await CreateIndexAsync(); @@ -220,19 +232,20 @@ namespace Store.Modules.CloudProduct.Services .RequestConfiguration(r => r.RequestTimeout(AdminTimeout)) ); - if (!bulkResponse.IsValid) + if (!bulkResponse.IsValid || bulkResponse.Errors) { throw new AppException("写入 OpenSearch 失败: " + bulkResponse.DebugInformation); } // 刷新索引，保证数据立即可搜索（仅用于演示，生产环境不建议频繁 Refresh） await RefreshIndexAsync(); + return list.Select(x => x.Id).Distinct().ToArray(); } /// /// ============ 3.2 根据数据库商品重建搜索索引 ============ /// - public async Task RebuildIndexAsync(IEnumerable goodsList) + public async Task RebuildIndexAsync(IEnumerable goodsList) { if (goodsList == null) throw new ArgumentNullException(nameof(goodsList)); @@ -240,6 +253,8 @@ namespace Store.Modules.CloudProduct.Services .Where(x => x != null && x.GoodsId > 0) .Select(x => new GoodsSearchDocument { + Platform = x.Platform, + ThumbUrl = x.ThumbUrl, Id = x.GoodsId, Title = x.Title ?? string.Empty, CatId1 = x.CatId1, @@ -251,10 +266,7 @@ namespace Store.Modules.CloudProduct.Services }) .ToList(); - if (documents.Count == 0) return 0; - - await BulkIndexGoodsAsync(documents); - return documents.Count; + return await BulkIndexGoodsAsync(documents); } /// @@ -269,7 +281,8 @@ namespace Store.Modules.CloudProduct.Services .RequestConfiguration(r => r.RequestTimeout(QueryTimeout))); if (!exists.Exists) { - return; + if (exists.ApiCall?.HttpStatusCode == 404) return; + throw new AppException("检查 OpenSearch 索引失败: " + exists.DebugInformation); } var response = await _client.BulkAsync(b => @@ -282,7 +295,9 @@ namespace Store.Modules.CloudProduct.Services return b; }); - if (!response.IsValid) + if (response.ApiCall?.HttpStatusCode is not (>= 200 and < 300) + || response.Items.Count != list.Count + || response.Items.Any(x => x.Status != 404 && (x.Error != null || x.Status < 200 || x.Status >= 300))) { throw new AppException("删除 OpenSearch 文档失败: " + response.DebugInformation); } @@ -296,6 +311,9 @@ namespace Store.Modules.CloudProduct.Services public async Task AddDocumentAsync(GoodsSearchDocument document) { if (document == null) throw new ArgumentNullException(nameof(document)); + await using var lease = await _indexLock.AcquireAsync(); + if (await _database.Select().AnyAsync(x => x.GoodsId == document.Id)) + throw new AppException("违规商品不能写入搜索索引"); var response = await _client.IndexAsync(document, i => i .Index(_indexName) @@ -522,6 +540,7 @@ namespace Store.Modules.CloudProduct.Services /// public async Task ClearIndexAsync() { + await using var lease = await _indexLock.AcquireAsync(); var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d .RequestConfiguration(r => r.RequestTimeout(QueryTimeout))); if (!existsResponse.Exists) @@ -539,7 +558,10 @@ namespace Store.Modules.CloudProduct.Services } Console.WriteLine($"索引 {_indexName} 已删除"); - return await CreateIndexAsync(); + var created = await CreateIndexAsync(); + await _database.Update().Set(x => x.IsSegmented, (byte)0) + .Where(x => x.Status == 0).ExecuteAffrowsAsync(); + return created; } } } diff --git a/Tests/CloudProduct.Checks/CloudProduct.Checks.csproj b/Tests/CloudProduct.Checks/CloudProduct.Checks.csproj new file mode 100644 index 0000000..828ba09 --- /dev/null +++ b/Tests/CloudProduct.Checks/CloudProduct.Checks.csproj @@ -0,0 +1,8 @@ + + + Exe + + + + + diff --git a/Tests/CloudProduct.Checks/IntegrationChecks.cs b/Tests/CloudProduct.Checks/IntegrationChecks.cs new file mode 100644 index 0000000..ffb1858 --- /dev/null +++ b/Tests/CloudProduct.Checks/IntegrationChecks.cs @@ -0,0 +1,329 @@ +using System.Reflection; +using System.Text; +using System.Text.Json; +using MAX.Framework.Commons; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging.Abstractions; +using OpenSearch.Client; +using OpenSearch.Net; +using Store.Modules.CloudProduct.Dtos; +using Store.Modules.CloudProduct.Entities; +using Store.Modules.CloudProduct.Repositorys; +using Store.Modules.CloudProduct.Services; + +internal static class IntegrationChecks +{ + public static async Task RunAsync(bool applyMigration = false, bool migrateViolationRows = false) + { + var root = Directory.GetCurrentDirectory(); + var config = new ConfigurationBuilder().SetBasePath(root) + .AddJsonFile("Store.API/appsettings.json") + .AddJsonFile("Store.API/appsettings.Development.json", optional: true) + .AddJsonFile("Modules/CloudProduct/appsettings.CloudProduct.json").Build(); + using var db = new FreeSql.FreeSqlBuilder() + .UseConnectionString(FreeSql.DataType.MySql, config["CloudProductConnection"]) + .UseAdoConnectionPool(true) + .Build(); + if (applyMigration) + { + var sql = await File.ReadAllTextAsync(Path.Combine(root, "Modules/CloudProduct/Sql/20261030_violation_reason_is_violation.sql")); + var present = Convert.ToInt32(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='violation_reason' AND column_name='is_violation'")); + if (present > 0) sql = sql[(sql.IndexOf(';') + 1)..]; + await db.Ado.ExecuteNonQueryAsync(sql); + Console.WriteLine("MIGRATION PASS: is_violation initialized"); + } + var run = Guid.NewGuid().ToString("N"); + var index = "violation-check-" + run; + var originalIndex = config["Search:IndexName"]!; + config["Search:IndexName"] = index; + config["Indexing:Mode"] = "Immediate"; + var client = new OpenSearchClient(new ConnectionSettings(new Uri(config["Search:Url"]!)).DefaultIndex(index)); + var type = typeof(IOpenSearchService).Assembly.GetType("Store.Modules.CloudProduct.Services.OpenSearchService")!; + var logger = Activator.CreateInstance(typeof(NullLogger<>).MakeGenericType(type)); + var gate = new GoodsIndexLock(config); + var search = (IOpenSearchService)Activator.CreateInstance(type, config, logger, db, gate)!; + GoodsViolationService Service(IOpenSearchService? target = null) => new( + new GoodsViolationRepository(db), gate, target ?? search, NullLogger.Instance); + var violations = Service(); + if (migrateViolationRows) + { + await using var lease = await gate.AcquireAsync(); + var invalid = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM goods_violation WHERE JSON_TYPE(reason_ids) <> 'ARRAY' OR JSON_LENGTH(reason_ids)=0")); + if (invalid != 0) throw new Exception("Migration requires nonempty reason arrays; inspect legacy records first"); + var expected = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(DISTINCT v.goods_id, r.reason_id) FROM goods_violation v JOIN JSON_TABLE(v.reason_ids, '$[*]' COLUMNS (reason_id BIGINT PATH '$' ERROR ON ERROR)) r ON TRUE")); + var migration = await File.ReadAllTextAsync(Path.Combine(root, "Modules/CloudProduct/Sql/20261030_goods_violation_reason_id.sql")); + var switchAt = migration.IndexOf("RENAME TABLE", StringComparison.Ordinal); + await db.Ado.ExecuteNonQueryAsync(migration[..switchAt]); + var copied = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM goods_violation_new")); + if (expected != copied) throw new Exception("Migration row count mismatch; original table not switched"); + await db.Ado.ExecuteNonQueryAsync(migration[switchAt..]); + Console.WriteLine($"MIGRATION PASS: {copied} reason rows migrated; original table retained as goods_violation_json_backup"); + } + var goodsType = type.Assembly.GetType("Store.Modules.CloudProduct.Services.GoodsService")!; + var goods = (IGoodsService)Activator.CreateInstance(goodsType, + new GoodsRepository(db), search, new NoRawData(), config, violations)!; + var first = 8_000_000_000_000_000L + Random.Shared.NextInt64(1_000_000_000); + long[] ids = [first, first + 1, first + 2, first + 3]; + // Exercise full-scan jobs against only this run's records in the shared test database. + db.GlobalFilter.Apply("integration-goods", x => ids.Contains(x.GoodsId)); + db.GlobalFilter.Apply("integration-violations", x => ids.Contains(x.GoodsId)); + var reasons = new List(); + string[] names = ["发布不合规信息", "出售假冒商品", "价格违规", "发布违禁信息"]; + var ignoredName = "其他原因-" + run; + string[] allowedNames = [ + "发布不合规信息", + "出售假冒商品", + "价格违规", + "发布违禁信息", + "材质夸大宣传", + "不当使用他人权利", + "发布虚假宣传信息", + "商品价格违规", + "出售盗版商品" + ]; + var ownsIds = false; + var checks = 0; + void Check(bool condition, string name) + { + if (!condition) throw new Exception("FAIL: " + name); + Console.WriteLine("PASS: " + name); + checks++; + } + async Task Exists(long id) + { + var response = await client.DocumentExistsAsync(id, d => d.Index(index)); + if (!response.IsValid && response.ApiCall?.HttpStatusCode != 404) + throw new Exception("Document existence request failed"); + return response.Exists; + } + async Task Reasons(long id) => (await db.Select() + .Where(x => x.GoodsId == id).OrderBy(x => x.ReasonId).ToListAsync(x => x.ReasonId)).ToArray(); + async Task Record(long id) => await db.Select() + .Where(x => x.GoodsId == id).FirstAsync(); + try + { + Check(await db.Select().CountAsync() >= 0, "read existing violation table"); + Check(await db.Select().CountAsync() >= 0, "read existing reason dictionary"); + Check(!await db.Select().AnyAsync(x => ids.Contains(x.GoodsId)) + && !await db.Select().AnyAsync(x => ids.Contains(x.GoodsId)), "test IDs are unused"); + ownsIds = true; + var enabledReasons = await db.Select().Where(x => x.IsViolation).ToListAsync(); + Check(enabledReasons.Select(x => x.Name).ToHashSet(StringComparer.Ordinal).SetEquals(allowedNames), "exactly the specified 9 dictionary reasons enabled"); + var existingReasonId = (await db.Select().Where(x => x.Name == names[0]).FirstAsync()).Id; + Check(await search.CreateIndexAsync(), "create isolated index on configured OpenSearch server"); + await search.BulkIndexGoodsAsync(ids.Select(id => new GoodsSearchDocument { Id = id, Title = "integration " + run })); + Check(await Exists(first), "initial product is indexed"); + var result = await violations.ReportAsync([ + new() { GoodsId = first, Reasons = " " + names[0] + " " }, + new() { GoodsId = first, Reasons = names[0] }, + new() { GoodsId = first, Reasons = names[1] }, + new() { GoodsId = first, Reasons = ignoredName } + ]); + Check(result.Count == 1 && result[0].GoodsId == first && result[0].IsOffShelf, "batch report and deduplication"); + Check(!await Exists(first), "report deletes real OpenSearch document"); + reasons.Add((await db.Select().Where(x => x.Name == names[0]).FirstAsync()).Id); + reasons.Add((await db.Select().Where(x => x.Name == names[1]).FirstAsync()).Id); + Check(reasons[0] == existingReasonId && reasons[1] > 0, "reuse enabled dictionary reason IDs"); + var record = await Record(first); + Check(record.IsIndexDeleted && (await Reasons(first)).Order().SequenceEqual(reasons.Take(2).Order()), "separate reasons and is_index_deleted persisted"); + await violations.ReportAsync([new() { GoodsId = first, Reasons = names[1] }, new() { GoodsId = first, Reasons = names[2] }]); + reasons.Add((await db.Select().Where(x => x.Name == names[2]).FirstAsync()).Id); + Check(await db.Select().Where(x => names.Contains(x.Name)).CountAsync() == 4, "repeated reasons reuse dictionary entries"); + record = await Record(first); + Check((await Reasons(first)).Order().SequenceEqual(reasons.Order()), "repeat report merges previous reasons"); + Check(await db.Select().Where(x => x.GoodsId == first).CountAsync() == 3, "one row per distinct product/reason pair"); + try + { + await db.Insert(await Record(first)).ExecuteAffrowsAsync(); + throw new Exception("Duplicate product/reason pair accepted"); + } + catch (Exception ex) when (ex.GetBaseException() is MySql.Data.MySqlClient.MySqlException { Number: 1062 }) + { + Check(true, "database composite primary key rejects duplicate product/reason pairs"); + } + await db.Update().Set(x => x.IsIndexDeleted, false) + .Where(x => x.GoodsId == first && x.ReasonId == reasons[0]).ExecuteAffrowsAsync(); + var violationRepository = new GoodsViolationRepository(db); + Check(!(await violationRepository.GetOffShelfGoodsIdsAsync([first])).Contains(first), + "one pending reason prevents reporting the product as fully off shelf"); + var pendingBatch = await violationRepository.GetBatchAsync(0, 1, true); + Check(pendingBatch.SequenceEqual(new[] { first }), "pending deletion pages distinct product IDs"); + await violationRepository.MarkDeletedAsync([first]); + var ignored = await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = ignoredName }]); + Check(ignored.Count == 1 && ignored[0].GoodsId == first + 1 && !ignored[0].IsOffShelf && await Record(first + 1) == null && await Exists(first + 1), + "unlisted reason records dictionary only and leaves index untouched"); + var disabledReport = await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = "品牌资质问题" }]); + Check(disabledReport.Count == 1 && disabledReport[0].GoodsId == first + 1 && !disabledReport[0].IsOffShelf + && await Record(first + 1) == null && await Exists(first + 1), "previously enabled reason outside the nine does not ban goods"); + Check(!((await db.Select().Where(x => x.Name == ignoredName).FirstAsync()).IsViolation), + "new dictionary reason defaults to not a violation"); + var before = await Reasons(first); + ignored = await violations.ReportAsync([new() { GoodsId = first, Reasons = ignoredName }]); + Check(ignored.Single(x => x.GoodsId == first).IsOffShelf && (await Reasons(first)).SequenceEqual(before) && !await Exists(first), + "ignored reason does not clear or alter existing violation"); + // A failing search transport proves dictionary-only reports never access OpenSearch. + + var status = await violations.GetStatusAsync([first, first + 1, first, first + 100]); + Check(status.Count == 3 && status[0].IsViolation && !status[1].IsViolation && !status[2].IsViolation, "batch status including missing goods"); + Check((await Reasons(first)).Length == 3, "scalar reasons stored in separate rows"); + Check((await violations.GetGoodsIdsByReasonAsync(reasons[2], first - 1)).Contains(first), "scalar reason query"); + try + { + await violations.ReportAsync([new() { GoodsId = first + 3, Reasons = " " }]); + throw new Exception("Invalid reason accepted"); + } + catch (AppException) { Check(await Record(first + 3) == null, "blank reason creates no row"); } + try + { + await search.AddDocumentAsync(new() { Id = first, Title = "blocked" }); + throw new Exception("Manual write accepted a violation"); + } + catch (AppException) { Check(!await Exists(first), "manual indexing rejects violation"); } + var indexed = await search.BulkIndexGoodsAsync([ + new() { Id = first, Title = "blocked" }, new() { Id = first + 1, Title = "allowed" } + ]); + Check(indexed.SequenceEqual(new[] { first + 1 }) && !await Exists(first) && await Exists(first + 1), "bulk indexing filters violations"); + + var imported = await goods.ImportGoodsAsync([ + new() { Goods = new() { GoodsId = first, Platform = 1, MinPrice = 0, MaxPrice = 0, CatId1 = 0, CatId2 = 0, CatId3 = 0, CatId4 = 0, ThumbUrl = "", Description = "", Title = "integration " + run } }, + new() { Goods = new() { GoodsId = first + 1, Platform = 1, MinPrice = 0, MaxPrice = 0, CatId1 = 0, CatId2 = 0, CatId3 = 0, CatId4 = 0, ThumbUrl = "", Description = "", Title = "integration " + run } } + ]); + Check(imported.SuccessCount == 2 && imported.FailedCount == 0 && !await Exists(first), "immediate import retains data but excludes violation: " + string.Join("; ", imported.Errors)); + // Inject a stale document to prove rebuild actively removes old violations. + var stale = await client.IndexAsync(new GoodsSearchDocument { Id = first, Title = "stale" }, i => i.Index(index).Id(first)); + Check(stale.IsValid, "seed stale violation document"); + var count = await goods.RebuildSearchIndexAsync([first, first + 1]); + Check(count == 1 && !await Exists(first) && await Exists(first + 1), "targeted rebuild removes stale violation and indexes eligible goods"); + + await db.Update().Set(x => x.IsSegmented, (byte)0) + .Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync(); + await search.DeleteDocumentsAsync([first + 1]); + Check(await goods.IndexPendingGoodsAsync(1) == 1 && await Exists(first + 1) && !await Exists(first), + "pending job filters violations before LIMIT 1"); + Check(await goods.RebuildSearchIndexAsync() == 1 && !await Exists(first), "full rebuild excludes violations"); + + // Real DB persistence with a deterministic failed search transport. + var failed = (IOpenSearchService)Activator.CreateInstance(type, config, logger, db, gate)!; + using var transport = new InMemoryConnection(Encoding.UTF8.GetBytes("{}"), 503); + var failedClient = new OpenSearchClient(new ConnectionSettings(new SingleNodeConnectionPool(new Uri("http://localhost:9200")), transport)); + type.GetField("_client", BindingFlags.Instance | BindingFlags.NonPublic)!.SetValue(failed, failedClient); + var oldTime = new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc); + await db.Update().Set(x => x.UpdatedTime, oldTime) + .Where(x => x.GoodsId == first).ExecuteAffrowsAsync(); + var unchangedBefore = await Record(first); + var unchangedReasons = await Reasons(first); + var unchanged = await Service(failed).ReportAsync([ + new() { GoodsId = first, Reasons = names[2] }, + new() { GoodsId = first, Reasons = " " + names[0] + " " }, + new() { GoodsId = first, Reasons = names[2] }, + new() { GoodsId = first, Reasons = ignoredName } + ]); + var unchangedAfter = await Record(first); + Check(unchanged.Count == 1 && unchanged[0].GoodsId == first && unchanged[0].IsOffShelf, + "duplicate/subset reasons skip unavailable OpenSearch and remain reported as off shelf"); + Check(unchangedBefore.UpdatedTime == unchangedAfter.UpdatedTime + && unchangedReasons.SequenceEqual(await Reasons(first)) && unchangedAfter.IsIndexDeleted, + "unchanged report preserves timestamp, reasons and deletion status"); + + var added = await Service(failed).ReportAsync([new() { GoodsId = first, Reasons = names[3] }]); + Check(!added.Single(x => x.GoodsId == first).IsOffShelf + && !(await Record(first)).IsIndexDeleted + && (await Reasons(first)).Length == 4, + "new valid reason updates record and attempts deletion again"); + var retry = await violations.ReportAsync([new() { GoodsId = first, Reasons = names[3] }]); + Check(retry.Single(x => x.GoodsId == first).IsOffShelf + && (await Record(first)).IsIndexDeleted, "unchanged reasons still retry a previously failed deletion"); + ignored = await Service(failed).ReportAsync([new() { GoodsId = first + 1, Reasons = ignoredName }]); + Check(!ignored.Single(x => x.GoodsId == first + 1).IsOffShelf, "ignored report does not require available OpenSearch"); + result = await Service(failed).ReportAsync([ + new() { GoodsId = first, Reasons = names[0] }, + new() { GoodsId = first + 2, Reasons = names[0] } + ]); + Check(result.Count == 2 && result[0].GoodsId == first && result[0].IsOffShelf + && result[1].GoodsId == first + 2 && !result[1].IsOffShelf, + "mixed batch separates unchanged successes from pending deletion"); + Check(!result.Single(x => x.GoodsId == first + 2).IsOffShelf && !(await Record(first + 2)).IsIndexDeleted, + "search failure persists pending deletion in real DB"); + await violations.DeleteViolationIndexesAsync(); + Check((await Record(first + 2)).IsIndexDeleted && !await Exists(first + 2), "retry deletes document and marks completed"); + + await Task.WhenAll( + search.BulkIndexGoodsAsync([new() { Id = first + 3, Title = "concurrent write" }]), + Service().ReportAsync([new() { GoodsId = first + 3, Reasons = names[0] }, new() { GoodsId = first + 3, Reasons = names[3] }]), + Service().ReportAsync([new() { GoodsId = first + 3, Reasons = names[1] }, new() { GoodsId = first + 3, Reasons = names[3] }])); + Check((await Reasons(first + 3)).Length == 3, + "concurrent reports preserve both reasons"); + Check(await db.Select().Where(x => x.Name == names[3]).CountAsync() == 1, "concurrent report reuses dictionary entry"); + Check(!await Exists(first + 3), "concurrent indexing cannot restore reported violation"); + + var removed = await violations.RemoveAsync([first, first, first + 2]); + Check(removed.RemovedCount == 2 && removed.RestoredGoodsIds.SequenceEqual(new[] { first }) + && removed.SkippedGoodsIds.SequenceEqual(new[] { first + 2 }), "remove duplicates and skip goods absent from database"); + Check(await Record(first) == null && await Exists(first), "remove violation restores real index"); + Check(!(await violations.GetStatusAsync([first]))[0].IsViolation, "restored product is no longer a violation"); + Check(await db.Select().Where(x => names.Contains(x.Name)).CountAsync() == 4, + "removal preserves shared reason dictionary"); + removed = await violations.RemoveAsync([first]); + Check(removed.RemovedCount == 0 && removed.RestoredGoodsIds.Contains(first), "repeat removal is idempotent"); + + await violations.ReportAsync([new() { GoodsId = first, Reasons = names[0] }]); + removed = await Service(failed).RemoveAsync([first]); + Check(removed.PendingRestoreGoodsIds.Contains(first) && await Record(first) == null + && await db.Select().AnyAsync(x => x.GoodsId == first && x.IsSegmented == 0), + "failed restore commits removal and leaves durable pending index flag"); + Check(await goods.IndexPendingGoodsAsync(1000) == 1 && await Exists(first), "pending job retries failed restoration"); + + await db.Update().Set(x => x.Status, (byte)1).Where(x => x.GoodsId == first + 1).ExecuteAffrowsAsync(); + await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = names[0] }]); + removed = await violations.RemoveAsync([first + 1]); + Check(removed.SkippedGoodsIds.Contains(first + 1) && await Record(first + 1) == null && !await Exists(first + 1), + "soft-deleted product is unblocked without resurrecting index"); + + await Task.WhenAll(violations.RemoveAsync([first]), + Service().ReportAsync([new() { GoodsId = first, Reasons = names[1] }])); + Check(((await Record(first)) != null) != await Exists(first), "concurrent removal and reporting leave consistent index state"); + var allReasonsResult = await violations.ReportAsync(allowedNames.Select(name => + new ReportGoodsViolationItem { GoodsId = first + 3, Reasons = name }).ToArray()); + Check(allReasonsResult.Count == 1 && allReasonsResult[0].GoodsId == first + 3 && allReasonsResult[0].IsOffShelf + && (await Reasons(first + 3)).ToHashSet() + .SetEquals(enabledReasons.Select(x => x.Id)), "all 9 reasons enter the product violation record"); + Console.WriteLine($"INTEGRATION PASS: {checks} checks"); + } + finally + { + // Only remove this run's isolated index and uniquely identified DB records. + var deleted = await client.Indices.DeleteAsync(index); + if (!deleted.IsValid && deleted.ApiCall?.HttpStatusCode != 404) + throw new Exception("Cleanup failed for test index: " + index); + // A separately running index job may have picked up test goods in the configured index. + var originalClient = new OpenSearchClient(new ConnectionSettings(new Uri(config["Search:Url"]!)).DefaultIndex(originalIndex)); + var existing = await originalClient.Indices.ExistsAsync(originalIndex); + if (existing.Exists && ownsIds) + { + var cleanup = await originalClient.BulkAsync(b => b.Index(originalIndex).DeleteMany(ids.Select(id => new GoodsSearchDocument { Id = id }))); + if (cleanup.ApiCall?.HttpStatusCode is not (>= 200 and < 300) + || cleanup.Items.Count != ids.Length + || cleanup.Items.Any(x => x.Status != 404 && (x.Status < 200 || x.Status >= 300))) + throw new Exception("Cleanup failed for test documents in configured index"); + await originalClient.Indices.RefreshAsync(originalIndex); + } + if (ownsIds) + { + await db.Delete().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync(); + await db.Delete().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync(); + } + await db.Delete().Where(x => x.Name == ignoredName).ExecuteAffrowsAsync(); + Check(!await db.Select().AnyAsync(x => ids.Contains(x.GoodsId)) + && !await db.Select().AnyAsync(x => ids.Contains(x.GoodsId)) + && !await db.Select().AnyAsync(x => x.Name == ignoredName), "test database records cleaned"); + Console.WriteLine("CLEANUP PASS: test index and records removed"); + } + } + + private sealed class NoRawData : IRawDataStorage + { + public bool Exists(long id) => false; + public Task ReadAsync(long id) => Task.FromResult(null); + public Task SaveAsync(long id, string raw) => throw new InvalidOperationException("Integration test does not write raw files"); + } +} diff --git a/Tests/CloudProduct.Checks/Program.cs b/Tests/CloudProduct.Checks/Program.cs new file mode 100644 index 0000000..f75b1cc --- /dev/null +++ b/Tests/CloudProduct.Checks/Program.cs @@ -0,0 +1,118 @@ +using System.Reflection; +using System.Text; +using MAX.Framework.Commons; +using System.Text.Json; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging.Abstractions; +using OpenSearch.Client; +using OpenSearch.Net; +using Store.Modules.CloudProduct.Controllers; +using Store.Modules.CloudProduct.Dtos; +using Store.Modules.CloudProduct.Entities; +using Store.Modules.CloudProduct.Services; + +if (args.Contains("--integration")) +{ + await IntegrationChecks.RunAsync(args.Contains("--apply-migration"), args.Contains("--migrate-violation-rows")); + return; +} + +var checks = 0; +void Check(bool condition, string message) +{ + if (!condition) throw new Exception(message); + checks++; +} +void Reject(Action action) +{ + try { action(); } + catch (AppException) { checks++; return; } + throw new Exception("Expected validation failure"); +} +var grouped = GoodsViolationService.ValidateAndGroup([ + new() { GoodsId = 12, Reasons = " 图片违规 " }, + new() { GoodsId = 12, Reasons = "描述违规" }, + new() { GoodsId = 12, Reasons = "图片违规" }, + new() { GoodsId = 12, Reasons = "涉嫌侵权" }, + new() { GoodsId = 12, Reasons = "描述违规" }, + new() { GoodsId = 13, Reasons = "中文原因" } +]); +Check(grouped.Count == 2 && grouped[12].ToHashSet().SetEquals(new[] { "图片违规", "描述违规", "涉嫌侵权" }), "Duplicate goods/reasons must merge"); +Check(grouped[13][0] == "中文原因", "Chinese reason text must survive"); +Reject(() => GoodsViolationService.ValidateAndGroup(null)); +Reject(() => GoodsViolationService.ValidateAndGroup([])); +Reject(() => GoodsViolationService.ValidateAndGroup([null!])); +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = -1, Reasons = "违规" }])); +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = 1, Reasons = "" }])); +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = 1, Reasons = " " }])); +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = 1, Reasons = null! }])); +Reject(() => GoodsViolationService.ValidateAndGroup(new ReportGoodsViolationItem[1001])); + +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = 1, Reasons = new string('中', 101) }])); +Reject(() => GoodsViolationService.ValidateAndGroup([new() { GoodsId = 1, Reasons = null! }])); + +// SQL generation only: no database connection or production data access. +using var db = new FreeSql.FreeSqlBuilder().UseConnectionString(FreeSql.DataType.MySql, + "Server=127.0.0.1;Database=checks;User Id=unused;Password=unused").Build(); +var sql = db.Select().As("g") + .Where(x => x.Status == 0 && x.IsSegmented == 0) + .Where("NOT EXISTS (SELECT 1 FROM goods_violation v WHERE v.goods_id = g.goods_id)") + .OrderBy(x => x.GoodsId).Limit(1000).ToSql(); +Check(sql.Contains("NOT EXISTS") && sql.IndexOf("NOT EXISTS") < sql.IndexOf("limit", StringComparison.OrdinalIgnoreCase), + "Violation filter must precede paging"); +Check(sql.Contains("FROM `goods` g", StringComparison.OrdinalIgnoreCase), "Anti-join alias must match: " + sql); +var pendingSql = db.Select().Where(x => !x.IsIndexDeleted).ToSql(); +Check(pendingSql.Contains("is_index_deleted") && !pendingSql.Contains("off_shelf_status"), "Renamed column mapping"); +var insertSql = db.Insert(new GoodsViolationEntity { GoodsId = 1, ReasonId = 2 }).NoneParameter().ToSql(); +Check(insertSql.Contains("reason_id") && !insertSql.Contains("reason_ids"), "Reason must be stored as a scalar ID"); + +var request = JsonSerializer.Deserialize("""{"goodsId":1,"reasons":"图片违规"}""", new JsonSerializerOptions(JsonSerializerDefaults.Web))!; +Check(request.Reasons == "图片违规", "JSON request accepts reason names"); +var reportJson = JsonSerializer.Serialize(new ReportGoodsViolationsResult { GoodsId = 1, IsOffShelf = true }, + new JsonSerializerOptions(JsonSerializerDefaults.Web)); +Check(reportJson == """{"goodsId":1,"isOffShelf":true}""", "Report response contains only goods ID and off-shelf status"); +try +{ + JsonSerializer.Deserialize("""{"goodsId":1,"reasons":["图片违规"]}""", new JsonSerializerOptions(JsonSerializerDefaults.Web)); + throw new Exception("Array reason accepted"); +} +catch (JsonException) { checks++; } +Check(!typeof(GoodsController).GetMethods().Any(x => x.Name.Contains("ByReason")), "Reason query must not be exposed"); +var removalService = new GoodsViolationService(null!, null!, null!, NullLogger.Instance); +Reject(() => removalService.RemoveAsync(null).GetAwaiter().GetResult()); +Reject(() => removalService.RemoveAsync([]).GetAwaiter().GetResult()); +Reject(() => removalService.RemoveAsync([0]).GetAwaiter().GetResult()); +Reject(() => removalService.RemoveAsync([-1]).GetAwaiter().GetResult()); +Reject(() => removalService.RemoveAsync(new long[1001]).GetAwaiter().GetResult()); + +// Exercise deletion with an in-memory OpenSearch transport, without a real server. +var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary +{ + ["Search:Url"] = "http://localhost:9200", ["Search:IndexName"] = "goods" +}).Build(); +var serviceType = typeof(IOpenSearchService).Assembly.GetType("Store.Modules.CloudProduct.Services.OpenSearchService")!; +var logger = Activator.CreateInstance(typeof(NullLogger<>).MakeGenericType(serviceType)); +async Task TestDelete(int status, string body, bool shouldFail) +{ + var service = (IOpenSearchService)Activator.CreateInstance(serviceType, config, logger, db, new GoodsIndexLock(config))!; + using var transport = new InMemoryConnection(Encoding.UTF8.GetBytes(body), status); + var client = new OpenSearchClient(new ConnectionSettings(new SingleNodeConnectionPool(new Uri("http://localhost:9200")), transport)); + serviceType.GetField("_client", BindingFlags.Instance | BindingFlags.NonPublic)!.SetValue(service, client); + try + { + await service.DeleteDocumentsAsync([123]); + Check(!shouldFail, $"HTTP {status}: expected delete failure"); + } + catch (AppException) + { + Check(shouldFail, $"HTTP {status}: unexpected delete failure"); + } +} +await TestDelete(403, "{}", true); +await TestDelete(500, "{}", true); +await TestDelete(404, "{}", false); +await TestDelete(200, """{"errors":false,"items":[{"delete":{"_index":"goods","_id":"123","status":200}}]}""", false); +await TestDelete(200, """{"errors":true,"items":[{"delete":{"_index":"goods","_id":"123","status":404,"result":"not_found"}}]}""", false); +await TestDelete(200, """{"errors":true,"items":[{"delete":{"_index":"goods","_id":"123","status":429,"error":{"type":"rejected","reason":"busy"}}}]}""", true); +await TestDelete(200, """{"errors":false,"items":[]}""", true); +Console.WriteLine($"PASS: {checks} checks (no external services accessed)");