云商品库 增加违规记录和违规字典

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<GoodsIndexJob>();
+			services.AddTransient<GoodsViolationIndexJob>();
+			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<IRecurringJobManager>().AddOrUpdate<GoodsViolationIndexJob>(
+					"goods-violation-delete-pending", job => job.ProcessPendingAsync(), "* * * * *");
 			}
 			catch (Exception ex)
 			{

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
 {
+	/// <summary>云商品管理：商品查询、导入、违规上报及搜索索引维护。</summary>
+	[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;
 		}
 
+		/// <summary>获取全部违规原因字典，包含未启用违规判定的原因，按ID升序返回。</summary>
 		[HttpGet]
+		[EndpointSummary("获取违规原因字典")]
+		public Task<List<ViolationReasonDto>> GetViolationReasons()
+			=> _violations.GetReasonsAsync();
+
+		/// <summary>批量上报违规商品；重复上报合并原因。</summary>
+		[HttpPost]
+		[EndpointSummary("批量上报违规商品")]
+		public Task<List<ReportGoodsViolationsResult>> ReportViolations([FromBody] ReportGoodsViolationItem[] items)
+			=> _violations.ReportAsync(items);
+
+		/// <summary>批量解除商品违规，恢复正常商品索引；失败后由后台索引任务重试。</summary>
+		[HttpPost]
+		[EndpointSummary("解除商品违规并恢复索引")]
+		public Task<RemoveGoodsViolationsResult> RemoveViolations([FromBody] long[] ids)
+			=> _violations.RemoveAsync(ids);
+
+		/// <summary>按商品编号批量查询是否违规。</summary>
+		[HttpPost]
+		[EndpointSummary("批量查询商品违规状态")]
+		public Task<List<GoodsViolationStatus>> GetViolationStatus([FromBody] long[] ids)
+			=> _violations.GetStatusAsync(ids);
+
+		/// <summary>获取服务器当前日期（长日期格式）。</summary>
+		[HttpGet]
+		[EndpointSummary("获取服务器当前日期")]
 		public async Task<string> Time()
 		{
 			return DateTime.Now.ToLongDateString();
 		}
+		/// <summary>手动执行待处理商品的索引任务。</summary>
 		[HttpGet]
+		[EndpointSummary("处理待建立索引的商品")]
 		public async Task Job()
 		{
 			await _goodsIndexJob.ProcessPendingAsync();
 		}
 		/// <summary>
-		/// 搜索
+		/// 分页搜索商品，可按标题、价格、类目、销量和平台筛选。
 		/// </summary>
-		/// <param name="request"></param>
-		/// <returns></returns>
+		/// <param name="request">商品筛选条件及分页参数。</param>
+		/// <returns>符合条件的商品列表及总数。</returns>
 		[HttpPost]
+		[EndpointSummary("分页搜索商品")]
 		public async Task<PageData<GoodsEntity>> SearchGoods(GoodsSearchRequest request)
 		{
 			// 1. 参数基础校验
@@ -145,6 +179,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// </summary>
 		[HttpGet]
 		[ActionName("SearchGoodsToIds")]
+		[EndpointSummary("按关键词搜索商品摘要")]
 		public async Task<PageData<GoodsSearchDocument>> 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 并走完整解析 + 入库 + 分词流程。
 		/// </summary>
 		[HttpGet]
+		[EndpointSummary("导入测试商品数据")]
 		public async Task<GoodsImportResult> ImportTestData()
 		{
 			var filePath = GetDataFilePath();
@@ -198,6 +234,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 导入接口：接收客户端批量上传的 JSON 文件，解析后入库并写入分词索引。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("批量导入商品 JSON 文件")]
 		public async Task<GoodsImportResult> ImportFiles([FromForm] List<IFormFile> files)
 		{
 			if (files == null || files.Count == 0)
@@ -247,6 +284,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 查询 OpenSearch 状态。
 		/// </summary>
 		[HttpGet]
+		[EndpointSummary("查询 OpenSearch 服务状态")]
 		public async Task<OpenSearchStatusDto> GetOpenSearchStatus()
 		{
 			return await _searchService.GetStatusAsync();
@@ -256,6 +294,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 检查搜索索引是否存在。
 		/// </summary>
 		[HttpGet]
+		[EndpointSummary("检查商品搜索索引是否存在")]
 		public async Task<bool> IndexExists()
 		{
 			return await _searchService.IndexExistsAsync();
@@ -265,6 +304,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 检查索引映射（分析器）是否符合当前服务要求。
 		/// </summary>
 		[HttpGet]
+		[EndpointSummary("检查商品索引映射是否正常")]
 		public async Task<bool> IndexMappingHealthy()
 		{
 			return await _searchService.IsIndexMappingHealthyAsync();
@@ -274,6 +314,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 创建搜索索引（含 IK 分词映射，已存在且映射正常时跳过）。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("创建商品搜索索引")]
 		public async Task<bool> CreateIndex()
 		{
 			return await _searchService.CreateIndexAsync();
@@ -283,6 +324,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 删除搜索索引。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("删除商品搜索索引")]
 		public async Task DeleteIndex()
 		{
 			await _searchService.DeleteIndexAsync();
@@ -292,6 +334,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 手动刷新搜索索引，使最新写入立即可搜索。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("刷新商品搜索索引")]
 		public async Task RefreshIndex()
 		{
 			await _searchService.RefreshIndexAsync();
@@ -301,6 +344,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 手动新增或覆盖一条搜索文档。
 		/// </summary>
 		[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。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("查询商品详情")]
 		public async Task<JsonNode> GetGoodsDetail([FromQuery] long goodsId)
 		{
 			var raw = await GetRawDataOrThrowAsync(goodsId);
@@ -326,6 +371,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 读取指定商品的原始数据（解压缩后原样返回）。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("读取商品原始数据")]
 		public async Task<string> GetGoodsRawData([FromQuery] long goodsId)
 		{
 			return await GetRawDataOrThrowAsync(goodsId);
@@ -336,6 +382,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// ids 为空时重建全部商品。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("重建商品搜索文档")]
 		public async Task<long> RebuildIndex([FromBody] long[]? ids)
 		{
 			return await _goodsService.RebuildSearchIndexAsync(ids);
@@ -346,6 +393,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// 软删除将 Status 置 1，硬删除物理删除记录；两种方式都会删除 OpenSearch 中对应文档。
 		/// </summary>
 		[HttpPost]
+		[EndpointSummary("批量删除商品")]
 		public async Task<long> DeleteGoods([FromBody] long[]? ids)
 		{
 			if (ids == null || ids.Length == 0)
@@ -361,6 +409,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// </summary>
 		[HttpGet]
 		[ActionName("ClearSearch")]
+		[EndpointSummary("清空并重建商品搜索索引")]
 		public async Task<bool> ClearSearchIndex()
 		{
 			return await _searchService.ClearIndexAsync();
@@ -371,6 +420,7 @@ namespace Store.Modules.CloudProduct.Controllers
 		/// </summary>
 		[HttpGet]
 		[ActionName("GetSearchCount")]
+		[EndpointSummary("统计商品搜索文档数量")]
 		public async Task<long> GetSearchDocumentCount()
 		{
 			return await _searchService.CountDocumentsAsync();

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; }
+    /// <summary>原因名称，例如“出售假冒商品”；新名称默认仅记录字典，不判定违规。</summary>
+    public string Reasons { get; set; } = string.Empty;
+}
+
+public class ReportGoodsViolationsResult
+{
+    public long GoodsId { get; set; }
+    /// <summary>存在违规记录且已完成索引下架时为 true，否则为 false。</summary>
+    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; } = [];
+    /// <summary>商品不存在、已删除或恢复期间再次被上报违规的编号。</summary>
+    public long[] SkippedGoodsIds { get; set; } = [];
+}

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; }
+}

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; }
+}

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<GoodsViolationEntity>(fsql), IScopedDependency
+{
+    public Task<List<ViolationReasonDto>> GetReasonsAsync() => Orm.Select<ViolationReasonEntity>()
+        .OrderBy(x => x.Id).ToListAsync(x => new ViolationReasonDto
+        {
+            Name = x.Name,
+            IsViolation = x.IsViolation
+        });
+
+    public Task<List<long>> GetViolationIdsAsync(long[] ids) => Select
+        .Where(x => ids.Contains(x.GoodsId)).Distinct().ToListAsync(x => x.GoodsId);
+
+    public async Task<List<long>> 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<int> RemoveAndScheduleRestoreAsync(long[] ids)
+    {
+        using var unit = Orm.CreateUnitOfWork();
+        var violations = unit.GetRepository<GoodsViolationEntity>();
+        var goods = unit.GetRepository<GoodsEntity>();
+        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<List<GoodsEntity>> GetRestorableGoodsAsync(long[] ids) => Orm.Select<GoodsEntity>()
+        .Where(x => ids.Contains(x.GoodsId) && x.Status == 0).ToListAsync();
+
+    public Task<int> MarkRestoredAsync(long[] ids) => Orm.Update<GoodsEntity>()
+        .Set(x => x.IsSegmented, (byte)1)
+        .Where(x => ids.Contains(x.GoodsId) && x.Status == 0).ExecuteAffrowsAsync();
+
+    // 调用方持有 GoodsIndexLock，事务内合并原因，避免并发覆盖。
+    public async Task<long[]> MergeAsync(Dictionary<long, string[]> items)
+    {
+        using var unit = Orm.CreateUnitOfWork();
+        var repository = unit.GetRepository<GoodsViolationEntity>();
+        var reasonRepository = unit.GetRepository<ViolationReasonEntity>();
+        var reasonIds = new Dictionary<string, uint>(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<long>();
+        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<int> MarkDeletedAsync(long[] ids) => Orm.Update<GoodsViolationEntity>()
+        .Set(x => x.IsIndexDeleted, true)
+        .Set(x => x.UpdatedTime, DateTime.UtcNow)
+        .Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
+
+    public Task<List<long>> 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<List<long>> 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);
+}

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;
+
+/// <summary>跨实例串行化违规上报和商品索引写入，防止下架后被旧批次写回。</summary>
+public class GoodsIndexLock(IConfiguration configuration) : ISingletonDependency
+{
+    public async Task<IAsyncDisposable> 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;
+        }
+    }
+}

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<GoodsSearchDocument>();
-			var indexedIds = new List<long>();
 
 			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;
 		}
 
 		/// <summary>
@@ -308,22 +308,22 @@ namespace Store.Modules.CloudProduct.Services
 		/// </summary>
 		public async Task<long> RebuildSearchIndexAsync(IEnumerable<long>? ids = null)
 		{
-			List<GoodsEntity> 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;
 		}
 

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<GoodsViolationService>()
+            .DeleteViolationIndexesAsync();
+    }
+}

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<GoodsViolationService> logger) : IScopedDependency
+{
+    public Task<List<ViolationReasonDto>> GetReasonsAsync() => repository.GetReasonsAsync();
+
+    public static Dictionary<long, string[]> 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<List<ReportGoodsViolationsResult>> 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<List<GoodsViolationStatus>> 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<RemoveGoodsViolationsResult> 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<List<long>> 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];
+        }
+    }
+}

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<GoodsSearchDocument> documents);
+		Task<long[]> BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents);
 		Task<bool> ClearIndexAsync();
 		Task<long> CountDocumentsAsync();
 		Task<bool> CreateIndexAsync();
@@ -24,7 +25,7 @@ namespace Store.Modules.CloudProduct.Services
 		Task<bool> IndexExistsAsync();
 		Task<bool> IsIndexMappingHealthyAsync();
 		Task RefreshIndexAsync();
-		Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList);
+		Task<long[]> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList);
 		Task<(long Total, List<GoodsSearchDocument> 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<OpenSearchService> _logger;
+		private readonly IFreeSql _database;
+		private readonly GoodsIndexLock _indexLock;
 
-		public OpenSearchService(IConfiguration configuration,ILogger<OpenSearchService> logger)
+		public OpenSearchService(IConfiguration configuration,ILogger<OpenSearchService> logger,
+			[FromKeyedServices("CloudProductConnection")] IFreeSql database, GoodsIndexLock indexLock)
 		{
+			_database = database;
+			_indexLock = indexLock;
 			var config = configuration.GetSection("Search");
 			_nodeUrl = config.GetValue<string>("Url") ?? "";
 			_indexName = config.GetValue<string>("IndexName") ?? "";
@@ -204,12 +210,18 @@ namespace Store.Modules.CloudProduct.Services
 		/// <summary>
 		/// ============ 3.1 批量写入商品搜索文档（存在则覆盖） ============
 		/// </summary>
-		public async Task BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents)
+		public async Task<long[]> BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> 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<GoodsViolationEntity>()
+				.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();
 		}
 
 		/// <summary>
 		/// ============ 3.2 根据数据库商品重建搜索索引 ============
 		/// </summary>
-		public async Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList)
+		public async Task<long[]> RebuildIndexAsync(IEnumerable<GoodsEntity> 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);
 		}
 
 		/// <summary>
@@ -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<GoodsViolationEntity>().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
 		/// </summary>
 		public async Task<bool> 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<GoodsEntity>().Set(x => x.IsSegmented, (byte)0)
+				.Where(x => x.Status == 0).ExecuteAffrowsAsync();
+			return created;
 		}
 	}
 }

Tests/CloudProduct.Checks/CloudProduct.Checks.csproj

@@ -0,0 +1,8 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="../../Modules/CloudProduct/Store.CloudProduct.csproj" />
+  </ItemGroup>
+</Project>

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<GoodsViolationService>.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<GoodsEntity>("integration-goods", x => ids.Contains(x.GoodsId));
+        db.GlobalFilter.Apply<GoodsViolationEntity>("integration-violations", x => ids.Contains(x.GoodsId));
+        var reasons = new List<uint>();
+        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<bool> Exists(long id)
+        {
+            var response = await client.DocumentExistsAsync<GoodsSearchDocument>(id, d => d.Index(index));
+            if (!response.IsValid && response.ApiCall?.HttpStatusCode != 404)
+                throw new Exception("Document existence request failed");
+            return response.Exists;
+        }
+        async Task<uint[]> Reasons(long id) => (await db.Select<GoodsViolationEntity>()
+            .Where(x => x.GoodsId == id).OrderBy(x => x.ReasonId).ToListAsync(x => x.ReasonId)).ToArray();
+        async Task<GoodsViolationEntity> Record(long id) => await db.Select<GoodsViolationEntity>()
+            .Where(x => x.GoodsId == id).FirstAsync();
+        try
+        {
+            Check(await db.Select<GoodsViolationEntity>().CountAsync() >= 0, "read existing violation table");
+            Check(await db.Select<ViolationReasonEntity>().CountAsync() >= 0, "read existing reason dictionary");
+            Check(!await db.Select<GoodsEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
+                && !await db.Select<GoodsViolationEntity>().AnyAsync(x => ids.Contains(x.GoodsId)), "test IDs are unused");
+            ownsIds = true;
+            var enabledReasons = await db.Select<ViolationReasonEntity>().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<ViolationReasonEntity>().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<ViolationReasonEntity>().Where(x => x.Name == names[0]).FirstAsync()).Id);
+            reasons.Add((await db.Select<ViolationReasonEntity>().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<ViolationReasonEntity>().Where(x => x.Name == names[2]).FirstAsync()).Id);
+            Check(await db.Select<ViolationReasonEntity>().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<GoodsViolationEntity>().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<GoodsViolationEntity>().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<ViolationReasonEntity>().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<GoodsEntity>().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<GoodsViolationEntity>().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<ViolationReasonEntity>().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<ViolationReasonEntity>().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<GoodsEntity>().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<GoodsEntity>().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<GoodsSearchDocument>(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<GoodsEntity>().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
+                await db.Delete<GoodsViolationEntity>().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
+            }
+            await db.Delete<ViolationReasonEntity>().Where(x => x.Name == ignoredName).ExecuteAffrowsAsync();
+            Check(!await db.Select<GoodsViolationEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
+                && !await db.Select<GoodsEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
+                && !await db.Select<ViolationReasonEntity>().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<string?> ReadAsync(long id) => Task.FromResult<string?>(null);
+        public Task<string> SaveAsync(long id, string raw) => throw new InvalidOperationException("Integration test does not write raw files");
+    }
+}

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<GoodsEntity>().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<GoodsViolationEntity>().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<ReportGoodsViolationItem>("""{"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<ReportGoodsViolationItem>("""{"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<GoodsViolationService>.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<string, string?>
+{
+    ["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)");