Web_BLS_RCUAction_Server/bls-rcu-action-backend/tests/consumer_reliability.test.js

import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import { createKafkaConsumers } from '../src/kafka/consumer.js';
import kafka from 'kafka-node';

// Mock kafka-node
vi.mock('kafka-node', () => {
  return {
    ConsumerGroup: vi.fn(),
    default: { ConsumerGroup: vi.fn() }
  };
});

describe('Consumer Reliability', () => {
  let mockConsumer;
  let onMessage;
  let onError;
  let healthCheck;
  
  const kafkaConfig = {
    brokers: ['localhost:9092'],
    groupId: 'test-group',
    clientId: 'test-client',
    topic: 'test-topic',
    autoCommitIntervalMs: 5000
  };

  beforeEach(() => {
    vi.clearAllMocks();
    
    mockConsumer = {
      on: vi.fn(),
      commit: vi.fn(),
      pause: vi.fn(),
      resume: vi.fn(),
      close: vi.fn()
    };
    
    kafka.ConsumerGroup.mockImplementation(function() {
      return mockConsumer;
    });
    
    onMessage = vi.fn().mockResolvedValue(true);
    onError = vi.fn();
    healthCheck = {
      shouldPause: vi.fn().mockResolvedValue(false),
      check: vi.fn().mockResolvedValue(true)
    };
  });

  it('should initialize with autoCommit: false', () => {
    createKafkaConsumers({ kafkaConfig, onMessage, onError });
    expect(kafka.ConsumerGroup).toHaveBeenCalledWith(
      expect.objectContaining({ autoCommit: false }),
      expect.anything()
    );
  });

  it('should commit offset after successful message processing', async () => {
    createKafkaConsumers({ kafkaConfig, onMessage, onError });
    
    // Simulate 'message' event
    const message = { value: 'test' };
    const messageHandler = mockConsumer.on.mock.calls.find(call => call[0] === 'message')[1];
    
    await messageHandler(message);
    
    expect(onMessage).toHaveBeenCalledWith(message);
    expect(mockConsumer.commit).toHaveBeenCalled();
  });

  it('should NOT commit if processing fails and health check says pause', async () => {
    onMessage.mockRejectedValue(new Error('Fail'));
    healthCheck.shouldPause.mockResolvedValue(true);
    createKafkaConsumers({ kafkaConfig, onMessage, onError, healthCheck });
    
    const messageHandler = mockConsumer.on.mock.calls.find(call => call[0] === 'message')[1];
    await messageHandler({ value: 'test' });
    
    expect(mockConsumer.commit).not.toHaveBeenCalled();
    expect(onError).toHaveBeenCalled();
  });

  it('should commit if processing fails but health check says continue (Data Error)', async () => {
    onMessage.mockRejectedValue(new Error('Data Error'));
    healthCheck.shouldPause.mockResolvedValue(false); // Do not pause, it's just bad data
    
    createKafkaConsumers({ kafkaConfig, onMessage, onError, healthCheck });
    
    const messageHandler = mockConsumer.on.mock.calls.find(call => call[0] === 'message')[1];
    await messageHandler({ value: 'bad_data' });
    
    expect(mockConsumer.commit).toHaveBeenCalled(); // Should commit to move past bad data
    expect(onError).toHaveBeenCalled(); // Should still report error
  });

  it('should pause and enter recovery mode if healthCheck.shouldPause returns true', async () => {
    vi.useFakeTimers();
    
    onMessage.mockRejectedValue(new Error('DB Error'));
    healthCheck.shouldPause.mockResolvedValue(true);
    healthCheck.check.mockResolvedValueOnce(false).mockResolvedValueOnce(true); // Fail once, then succeed

    createKafkaConsumers({ kafkaConfig, onMessage, onError, healthCheck });
    const messageHandler = mockConsumer.on.mock.calls.find(call => call[0] === 'message')[1];
    
    // Trigger error
    await messageHandler({ value: 'fail' });

    expect(mockConsumer.pause).toHaveBeenCalled();
    expect(healthCheck.shouldPause).toHaveBeenCalled();
    
    // Fast-forward time for interval check (1st check - fails)
    await vi.advanceTimersByTimeAsync(60000); 
    expect(healthCheck.check).toHaveBeenCalledTimes(1);
    expect(mockConsumer.resume).not.toHaveBeenCalled(); 

    // Fast-forward time for interval check (2nd check - succeeds)
    await vi.advanceTimersByTimeAsync(60000); 
    expect(healthCheck.check).toHaveBeenCalledTimes(2);
    expect(mockConsumer.resume).toHaveBeenCalled();

    vi.useRealTimers();
  });
});