const mockProcessors = new Map<string, (job: any) => Promise<unknown>>();
const mockQueueProcess = jest.fn((name: string, ...rest: unknown[]) => {
  // Bull accepts process(name, handler) or process(name, concurrency, handler).
  const processor = rest[rest.length - 1] as (job: any) => Promise<unknown>;
  mockProcessors.set(name, processor);
});
const mockQueueAdd = jest.fn().mockResolvedValue(undefined);
const mockDeliverWebhook = jest.fn().mockResolvedValue({ success: true });
const mockTranslateStructured = jest.fn();
const mockLockProcessingJob = jest.fn().mockResolvedValue(undefined);
const mockLoggerError = jest.fn();

jest.mock('../../queue', () => ({
  translationQueue: {
    process: mockQueueProcess,
    add: mockQueueAdd,
    on: jest.fn(),
    close: jest.fn(),
    isReady: jest.fn().mockResolvedValue(true),
  },
}));
jest.mock('../../services/geminiClient', () => ({
  geminiClient: {
    translate: jest.fn(),
    translateBulk: jest.fn(),
    translateStructured: mockTranslateStructured,
  },
}));
jest.mock('../../services/webhookService', () => ({
  deliverWebhook: mockDeliverWebhook,
}));
jest.mock('../../utils/logger', () => ({
  logger: {
    info: jest.fn(),
    warn: jest.fn(),
    error: mockLoggerError,
  },
}));
jest.mock('../../utils/metrics', () => ({
  trackTranslationJob: jest.fn(),
  trackTokensProcessed: jest.fn(),
}));
jest.mock('../../workers/orphanDisputeReconciler', () => ({
  reconcilePermanentOrphanDisputes: jest.fn(),
}));
jest.mock('../../queue/bulkQueueDispatcher', () => ({
  configureBulkQueueDispatcher: jest.fn(),
  startBulkQueueDispatcher: jest.fn(),
  stopBulkQueueDispatcher: jest.fn(),
}));
jest.mock('../../workerLifecycle', () => ({
  startWorkerRuntime: jest.fn(),
  shutdownWorkerRuntime: jest.fn(),
}));
jest.mock('../../workerLock', () => ({
  JobCancellationWonError: class JobCancellationWonError extends Error {},
  lockProcessingJob: mockLockProcessingJob,
}));
jest.mock('../../workerHeartbeat', () => ({
  getWorkerHeartbeatKey: jest.fn(() => 'worker:test'),
  WorkerHeartbeat: class WorkerHeartbeat {
    start(): void {}
    stop(): Promise<void> { return Promise.resolve(); }
  },
}));
jest.mock('../../config', () => ({
  getRedisUrl: jest.fn(() => 'redis://localhost:6379'),
  initializeConfigFromDatabase: jest.fn(),
  startConfigSubscription: jest.fn(),
  useEnvironmentConfigFallback: jest.fn(),
}));

process.env.WORKER_ID = 'bulk-content-test-worker';

import { prismaMock } from '../setup';
import { registerProcessors } from '../../worker';
import { Tone } from '../../types';
import type { BulkContentTranslationJobData } from '../../types';

const jobData: BulkContentTranslationJobData = {
  jobId: 'job-1',
  clientJobId: 'wp_bulk_41',
  userId: 'user-1',
  items: [
    { ref: 'content:1:es', targetLang: 'es', title: 'Title', content: 'Body' },
    { ref: 'content:1:fr', targetLang: 'fr', title: 'Title', content: 'Body' },
  ],
  sourceLang: 'en',
  tone: Tone.NEUTRAL,
  callbackUrl: 'https://site.example.test/callback',
  callbackSecret: 'sixteen-character-secret',
};

const mockOutbox = {
  findMany: jest.fn().mockResolvedValue([]),
  updateMany: jest.fn().mockResolvedValue({ count: 0 }),
  create: jest.fn().mockResolvedValue({}),
};
const pendingOutboxEntries: any[] = [];

function processor(): (job: any) => Promise<any> {
  registerProcessors();
  const value = mockProcessors.get('bulk-content');
  if (!value) throw new Error('bulk-content processor was not registered');
  return value;
}

describe('bulk-content worker', () => {
  beforeEach(() => {
    mockProcessors.clear();
    mockQueueProcess.mockClear();
    mockDeliverWebhook.mockClear();
    mockTranslateStructured.mockReset();
    mockLockProcessingJob.mockClear();
    mockLoggerError.mockClear();
    pendingOutboxEntries.length = 0;
    mockOutbox.findMany.mockImplementation(async () => pendingOutboxEntries);
    mockOutbox.updateMany.mockResolvedValue({ count: 1 });
    mockOutbox.create.mockImplementation(async (args: any) => {
      const data = args.data;
      const entry = {
        ...data,
        attempts: 0,
        job: {
          callback_url: jobData.callbackUrl,
          callback_secret: jobData.callbackSecret,
        },
      };
      pendingOutboxEntries.push(entry);
      return entry;
    });
    Object.assign(prismaMock, { webhookOutbox: mockOutbox });
    (prismaMock.$transaction as jest.Mock).mockImplementation(
      async (callback: (tx: typeof prismaMock) => Promise<unknown>) => callback(prismaMock)
    );
    prismaMock.translationJob.updateMany.mockResolvedValue({ count: 1 } as never);
    prismaMock.creditTransaction.findFirst.mockResolvedValue({ balance_after: 1000 } as never);
    prismaMock.creditTransaction.create.mockResolvedValue({} as never);
    prismaMock.user.update.mockResolvedValue({} as never);
    prismaMock.translationJob.update.mockResolvedValue({} as never);
  });

  it('translates every item with one Gemini call each and delivers one webhook with per-item results', async () => {
    mockTranslateStructured
      .mockResolvedValueOnce({
        fields: { title: 'Título', content: 'Cuerpo' },
        translatedFields: { title: 'Título', content: 'Cuerpo' },
        tokens_used: 4, input_tokens: 2, output_tokens: 2, processing_time_ms: 5, model_used: 'gemini-3.1-flash-lite',
      })
      .mockResolvedValueOnce({
        fields: { title: 'Titre', content: 'Corps' },
        translatedFields: { title: 'Titre', content: 'Corps' },
        tokens_used: 4, input_tokens: 2, output_tokens: 2, processing_time_ms: 5, model_used: 'gemini-3.1-flash-lite',
      });

    const result = await processor()({ data: jobData, opts: { attempts: 1 }, attemptsMade: 0 });

    expect(mockTranslateStructured).toHaveBeenCalledTimes(2);
    expect(mockTranslateStructured).toHaveBeenNthCalledWith(1, { title: 'Title', content: 'Body' }, 'en', 'es', Tone.NEUTRAL);
    expect(mockTranslateStructured).toHaveBeenNthCalledWith(2, { title: 'Title', content: 'Body' }, 'en', 'fr', Tone.NEUTRAL);

    expect(mockOutbox.create).toHaveBeenCalledTimes(1);
    expect(mockDeliverWebhook).toHaveBeenCalledTimes(1);
    const payload = mockOutbox.create.mock.calls[0][0].data.payload;
    expect(payload.event).toBe('bulk_content_translation.completed');
    expect(payload.results).toEqual([
      { ref: 'content:1:es', status: 'completed', translatedTitle: 'Título', translatedExcerpt: undefined, translatedContent: 'Cuerpo' },
      { ref: 'content:1:fr', status: 'completed', translatedTitle: 'Titre', translatedExcerpt: undefined, translatedContent: 'Corps' },
    ]);
    expect(payload.failed_count).toBe(0);

    // Exactly one deduction for the whole batch.
    expect(prismaMock.creditTransaction.create).toHaveBeenCalledTimes(1);

    expect(result.totalFailedCount).toBe(0);
  });

  it('preserves canonical structured fields and submission identity in the completion handoff', async () => {
    const canonicalData: BulkContentTranslationJobData = {
      ...jobData,
      submissionId: '4d8f3f7e-2ae1-4f50-8cb7-4a6a7ca16de1',
      items: [{
        ref: 'content:1:es',
        targetLang: 'es',
        fields: { title: 'Title', seo_description: 'Description' },
      }],
    };
    mockTranslateStructured.mockResolvedValue({
      fields: { title: 'Título', seo_description: 'Descripción' },
      translatedFields: { title: 'Título', seo_description: 'Descripción' },
      tokens_used: 4, input_tokens: 2, output_tokens: 2, processing_time_ms: 5, model_used: 'gemini-3.1-flash-lite',
    });

    await processor()({ data: canonicalData, opts: { attempts: 1 }, attemptsMade: 0 });

    const payload = mockOutbox.create.mock.calls[0][0].data.payload;
    expect(payload.submission_id).toBe(canonicalData.submissionId);
    expect(payload.results).toEqual([{
      ref: 'content:1:es',
      status: 'completed',
      fields: { title: 'Título', seo_description: 'Descripción' },
    }]);
  });

  it('isolates a single item failure without aborting the batch, and bills only successes', async () => {
    mockTranslateStructured
      .mockRejectedValueOnce(new Error('Gemini timed out'))
      .mockResolvedValueOnce({
        fields: { title: 'Titre', content: 'Corps' },
        translatedFields: { title: 'Titre', content: 'Corps' },
        tokens_used: 4, input_tokens: 2, output_tokens: 2, processing_time_ms: 5, model_used: 'gemini-3.1-flash-lite',
      });

    const result = await processor()({ data: jobData, opts: { attempts: 1 }, attemptsMade: 0 });

    expect(mockTranslateStructured).toHaveBeenCalledTimes(2);
    const payload = mockOutbox.create.mock.calls[0][0].data.payload;
    expect(payload.results[0]).toEqual({ ref: 'content:1:es', status: 'failed', error: 'Gemini timed out' });
    expect(payload.results[1]).toMatchObject({ ref: 'content:1:fr', status: 'completed' });
    expect(payload.failed_count).toBe(1);

    // Billed characters must reflect only the successful item ("Title" + "Body" = 9 chars),
    // not the failed item — a failed item is retried by the client and must not be double-billed.
    const deductionCall = prismaMock.creditTransaction.create.mock.calls[0][0] as any;
    expect(deductionCall.data.amount).toBe(-9);
    expect(payload.total_characters_used).toBe(9);

    // Exactly one deduction and one webhook for the whole batch, despite the partial failure.
    expect(prismaMock.creditTransaction.create).toHaveBeenCalledTimes(1);
    expect(mockOutbox.create).toHaveBeenCalledTimes(1);
    expect(mockDeliverWebhook).toHaveBeenCalledTimes(1);
    expect(result.totalFailedCount).toBe(1);
  });

  it('replays a completed terminal job without provider or billing work', async () => {
    prismaMock.translationJob.updateMany.mockResolvedValue({ count: 0 } as never);
    prismaMock.translationJob.findUnique.mockResolvedValue({
      status: 'completed',
      translation: '[{"ref":"content:1:es","status":"completed"}]',
    } as never);

    const result = await processor()({ data: jobData, opts: { attempts: 1 }, attemptsMade: 0 });

    expect(result).toEqual({ translation: '[{"ref":"content:1:es","status":"completed"}]' });
    expect(mockTranslateStructured).not.toHaveBeenCalled();
    expect(prismaMock.creditTransaction.create).not.toHaveBeenCalled();
  });
});