import { CrmRadarBufferService } from '../../../src/modules/crm/service/buffer'; import { CrmTrackEntity } from '../../../src/modules/crm/entity/track'; describe('高并发雷达削峰缓冲服务 (CrmRadarBufferService)', () => { let bufferService: CrmRadarBufferService; let insertedBatches: Array; let mockTrackRepo: any; let mockInsertQb: any; beforeEach(() => { insertedBatches = []; mockInsertQb = { insert: jest.fn().mockReturnThis(), into: jest.fn().mockReturnThis(), values: jest.fn().mockImplementation((batch: any[]) => { insertedBatches.push([...batch]); return mockInsertQb; }), execute: jest.fn().mockResolvedValue({ identifiers: [] }), }; mockTrackRepo = { createQueryBuilder: jest.fn().mockReturnValue(mockInsertQb), save: jest.fn().mockImplementation(async (items: any[]) => { insertedBatches.push(Array.isArray(items) ? [...items] : [items]); return items; }), }; bufferService = new CrmRadarBufferService(); bufferService.crmTrackEntity = mockTrackRepo; // 单元测试中设置较小的批次阈值,禁用自动定时器 bufferService.setBatchSize(5); bufferService.setFlushInterval(0); // 禁用定时器以便手动可控测试 }); afterEach(() => { bufferService.stop(); }); describe('1. 缓冲入队与队列状态 (Enqueue & Queue Length)', () => { it('调用 enqueue 能够将足迹实体推入内存环形队列,且非阻塞瞬间返回', async () => { const track: Partial = { cardId: 101, tenantId: 88, visitorId: 'vis_test_001', touchType: 'nfc', country: 'United Arab Emirates', }; const queued = await bufferService.enqueue(track as CrmTrackEntity); expect(queued).toBe(true); expect(bufferService.getQueueLength()).toBe(1); // 队列未达批量阈值,尚未触发物理落盘 expect(insertedBatches).toHaveLength(0); }); it('队列到达批次阈值 (BatchSize) 时应自动触发批量 Bulk Insert 落盘', async () => { bufferService.setBatchSize(3); await bufferService.enqueue({ cardId: 1, visitorId: 'v1' } as any); await bufferService.enqueue({ cardId: 2, visitorId: 'v2' } as any); expect(insertedBatches).toHaveLength(0); // 放入第 3 条达到阈值,触发自动 flush await bufferService.enqueue({ cardId: 3, visitorId: 'v3' } as any); expect(insertedBatches).toHaveLength(1); expect(insertedBatches[0]).toHaveLength(3); expect(bufferService.getQueueLength()).toBe(0); }); }); describe('2. 手动批量落盘与关闭刷新 (Flush & Stop)', () => { it('显式调用 flush 应将队列中剩余所有足迹一次性批量入库', async () => { bufferService.setBatchSize(100); await bufferService.enqueue({ cardId: 1, visitorId: 'v1' } as any); await bufferService.enqueue({ cardId: 2, visitorId: 'v2' } as any); expect(bufferService.getQueueLength()).toBe(2); const flushedCount = await bufferService.flush(); expect(flushedCount).toBe(2); expect(bufferService.getQueueLength()).toBe(0); expect(insertedBatches).toHaveLength(1); expect(insertedBatches[0][0].visitorId).toBe('v1'); expect(insertedBatches[0][1].visitorId).toBe('v2'); }); it('空队列调用 flush 应安全返回 0 且不执行数据库写入', async () => { const flushedCount = await bufferService.flush(); expect(flushedCount).toBe(0); expect(insertedBatches).toHaveLength(0); }); it('调用 stop 应清除内部定时器并把剩余积攒数据完全 flush', async () => { await bufferService.enqueue({ cardId: 99, visitorId: 'v_exit' } as any); bufferService.stop(); expect(insertedBatches).toHaveLength(1); expect(bufferService.getQueueLength()).toBe(0); }); }); });