OmniCard/backend/test/modules/crm/radar-buffer.test.ts

import { CrmRadarBufferService } from '../../../src/modules/crm/service/buffer';
import { CrmTrackEntity } from '../../../src/modules/crm/entity/track';

describe('高并发雷达削峰缓冲服务 (CrmRadarBufferService)', () => {
  let bufferService: CrmRadarBufferService;
  let insertedBatches: Array<any[]>;
  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<CrmTrackEntity> = {
        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);
    });
  });
});