import { beforeEach, describe, expect, it, jest } from '@jest/globals';
import { Prisma } from '@prisma/client';

jest.mock('../../../queue', () => ({
  translationQueue: { add: jest.fn() },
}));
jest.mock('@prisma/client', () => ({
  Prisma: { DbNull: Symbol('db-null') },
}));
jest.mock('../../../utils/logger', () => ({
  logger: { error: jest.fn() },
}));
import {
  BulkQueuePayload,
  BulkQueueCandidate,
  BulkQueueDispatcherDependencies,
  createBulkQueueDispatcher,
  createPrismaBulkQueueDispatcher,
} from '../../../queue/bulkQueueDispatcher';
import crypto from 'crypto';

type StoredJob = BulkQueueCandidate & {
  queue_payload: Record<string, unknown> | null;
  callback_secret: string | null;
};

const payload = {
  type: 'bulk-strings',
  jobId: 'job-1',
  clientJobId: 'wp-1',
  userId: 'user-1',
  plugin: 'international',
  strings: [{ id: 'title', content: 'Hello' }],
  sourceLang: 'en',
  targetLangs: ['es', 'fr'],
  tone: 'neutral',
  callbackUrl: 'https://site.example.test/callback',
};

function dependencies(
  job: StoredJob,
  add?: (name: string, payload: BulkQueuePayload, options: { jobId: string }) => Promise<unknown>
):
  BulkQueueDispatcherDependencies {
  const resolvedAdd = add ?? jest.fn<BulkQueueDispatcherDependencies['add']>().mockResolvedValue({ id: job.id });
  return {
    listCandidates: jest.fn<BulkQueueDispatcherDependencies['listCandidates']>(async () => [job]),
    claim: jest.fn(async (id: string, leaseId: string, now: Date) => {
      if (id !== job.id || job.status !== 'pending' || job.queued_at ||
        (job.queue_lease_expires_at && job.queue_lease_expires_at >= now)) {
        return null;
      }
      job.queue_lease_id = leaseId;
      job.queue_lease_expires_at = new Date(now.getTime() + 120_000);
      return { callbackSecret: job.callback_secret };
    }),
    markQueued: jest.fn(async (id: string, leaseId: string) => {
      if (id !== job.id || job.queue_lease_id !== leaseId || job.queued_at) return false;
      job.queued_at = new Date();
      job.queue_lease_id = null;
      job.queue_lease_expires_at = null;
      return true;
    }),
    release: jest.fn(async (id: string, leaseId: string) => {
      if (id === job.id && job.queue_lease_id === leaseId && !job.queued_at) {
        job.queue_lease_id = null;
        job.queue_lease_expires_at = null;
      }
    }),
    add: resolvedAdd,
    createLeaseId: (() => {
      let value = 0;
      return () => `lease-${++value}`;
    })(),
    now: () => new Date('2026-08-02T00:00:00.000Z'),
  };
}

function pendingJob(): StoredJob {
  return {
    id: 'job-1',
    status: 'pending',
    queued_at: null,
    queue_lease_id: null,
    queue_lease_expires_at: null,
    queue_payload: payload,
    callback_secret: 'sixteen-character-secret',
    created_at: new Date('2026-08-01T00:00:00.000Z'),
  };
}

beforeEach(() => {
  jest.clearAllMocks();
});

describe('bulk queue dispatcher', () => {
  it('excludes every database-null payload batch and immediately queues a later persisted payload', async () => {
    const nullPayloadRows = Array.from({ length: 201 }, (_, index) => ({
      ...pendingJob(), id: `legacy-${index.toString().padStart(3, '0')}`, queue_payload: null,
    }));
    const valid = pendingJob();
    valid.id = 'valid';
    valid.queue_payload = { ...payload, jobId: valid.id };
    const delegate = {
      findMany: jest.fn(async (args: {
        where: { queue_payload?: { not?: unknown } }; cursor?: { id: string }; take: number;
      }) => {
        const rows = args.where.queue_payload?.not === Prisma.DbNull ? [valid] : nullPayloadRows;
        const start = args.cursor ? rows.findIndex((row) => row.id === args.cursor!.id) + 1 : 0;
        return rows.slice(start, start + args.take);
      }),
      findFirst: jest.fn(async () => ({ callback_secret: valid.callback_secret })),
      updateMany: jest.fn(async () => ({ count: 1 })),
    };

    await createPrismaBulkQueueDispatcher(delegate).drain();

    expect(delegate.findMany).toHaveBeenCalledTimes(1);
    const firstQuery = (delegate.findMany as jest.Mock).mock.calls[0][0] as {
      where: { queue_payload?: { not?: unknown } };
    };
    expect(firstQuery.where.queue_payload).toEqual({ not: Prisma.DbNull });
    expect(require('../../../queue').translationQueue.add).toHaveBeenCalledWith(
      'bulk-strings', { ...valid.queue_payload, callbackSecret: valid.callback_secret }, { jobId: valid.id }
    );
  });

  it('drains a committed job after submission dies before its first enqueue', async () => {
    const job = pendingJob();
    const deps = dependencies(job);

    await createBulkQueueDispatcher(deps).drain();

    expect(deps.add).toHaveBeenCalledWith('bulk-strings', { ...payload, callbackSecret: job.callback_secret }, { jobId: job.id });
    expect(job.queued_at).toBeInstanceOf(Date);
  });

  it('starts immediately, schedules bounded recurring drains, and stops cleanly', async () => {
    jest.useFakeTimers();
    const job = pendingJob();
    const deps = dependencies(job);
    const dispatcher = createBulkQueueDispatcher(deps);

    await dispatcher.start();
    await jest.advanceTimersByTimeAsync(5_000);
    await dispatcher.stop();

    expect(deps.add).toHaveBeenCalledTimes(1);
    expect(deps.listCandidates).toHaveBeenCalledTimes(2);
    jest.useRealTimers();
  });

  it('claims concurrently drained rows once', async () => {
    const job = pendingJob();
    const deps = dependencies(job);
    const first = createBulkQueueDispatcher(deps);
    const second = createBulkQueueDispatcher(deps);

    await Promise.all([first.drain(), second.drain()]);

    expect(deps.add).toHaveBeenCalledTimes(1);
    expect(deps.add).toHaveBeenCalledWith('bulk-strings', { ...payload, callbackSecret: job.callback_secret }, { jobId: job.id });
  });

  it('recovers an expired post-add lease with Bull stable jobId', async () => {
    const job = pendingJob();
    job.queue_lease_id = 'dead-process';
    job.queue_lease_expires_at = new Date('2026-08-01T23:59:59.999Z');
    const bullJobIds = new Set([job.id]);
    const add = jest.fn<BulkQueueDispatcherDependencies['add']>(async (_name, _payload, options) => {
      bullJobIds.add(options.jobId);
      return { id: options.jobId };
    });
    const deps = dependencies(job, add);

    await createBulkQueueDispatcher(deps).drain();

    expect(deps.add).toHaveBeenCalledWith('bulk-strings', { ...payload, callbackSecret: job.callback_secret }, { jobId: job.id });
    expect(bullJobIds).toEqual(new Set([job.id]));
    expect(job.queued_at).toBeInstanceOf(Date);
  });

  it('releases a failed enqueue for bounded recurring retry', async () => {
    const job = pendingJob();
    const add = jest.fn<(name: string, payload: BulkQueuePayload, options: { jobId: string }) => Promise<unknown>>()
      .mockRejectedValueOnce(new Error('Redis unavailable'))
      .mockResolvedValueOnce({ id: job.id });
    const deps = dependencies(job, add);
    const dispatcher = createBulkQueueDispatcher(deps);

    await expect(dispatcher.drain()).resolves.toMatchObject({ failed: 1 });
    expect(job.queue_lease_id).toBeNull();
    await dispatcher.drain();

    expect(add).toHaveBeenCalledTimes(2);
    expect(job.queued_at).toBeInstanceOf(Date);
  });

  it.each(['cancelled', 'processing', 'completed', 'failed'])('skips %s rows', async (status) => {
    const job = pendingJob();
    job.status = status;
    const deps = dependencies(job);

    await createBulkQueueDispatcher(deps).drain();

    expect(deps.add).not.toHaveBeenCalled();
  });

  it('keeps the callback secret out of stored payload and passes exact secret to Bull', async () => {
    const job = pendingJob();
    const deps = dependencies(job);

    await createBulkQueueDispatcher(deps).drain();

    const bullPayload = (deps.add as jest.Mock).mock.calls[0][1] as BulkQueuePayload;
    expect(JSON.stringify(job.queue_payload)).not.toContain(job.callback_secret!);
    expect(bullPayload.callbackSecret).toBe(job.callback_secret);
    const body = '{"event":"bulk_translation.completed"}';
    expect(crypto.createHmac('sha256', bullPayload.callbackSecret).update(`1.${body}`).digest('hex'))
      .toBe(crypto.createHmac('sha256', job.callback_secret!).update(`1.${body}`).digest('hex'));
  });

  it('maps Prisma claimed-row callback_secret to Bull callbackSecret after atomic claim', async () => {
    const job = pendingJob();
    type ClaimedRow = Prisma.TranslationJobGetPayload<{ select: { callback_secret: true } }>;
    const claimedRow: ClaimedRow = { callback_secret: job.callback_secret };
    const delegate = {
      findMany: jest.fn(async () => [job]),
      findFirst: jest.fn(async () => claimedRow),
      updateMany: jest.fn(async () => ({ count: 1 })),
    };

    await createPrismaBulkQueueDispatcher(delegate).drain();

    const updateCalls = (delegate.updateMany as jest.Mock).mock.calls as Array<[{ data: Record<string, unknown> }]>;
    const claimCall = updateCalls.find(([args]) => args.data.queue_lease_id);
    expect(claimCall).toBeDefined();
    expect(delegate.findFirst).toHaveBeenCalledTimes(1);
    expect(((delegate.findMany as jest.Mock).mock.calls as Array<[{ select: Record<string, unknown> }]>).every(([args]) => !args.select.callback_secret)).toBe(true);
    expect(require('../../../queue').translationQueue.add).toHaveBeenCalledTimes(1);
    expect(require('../../../queue').translationQueue.add).toHaveBeenCalledWith(
      'bulk-strings', { ...payload, callbackSecret: job.callback_secret }, { jobId: job.id }
    );
    const addOrder = (require('../../../queue').translationQueue.add as jest.Mock).mock.invocationCallOrder[0];
    const queuedOrder = (delegate.updateMany as jest.Mock).mock.invocationCallOrder[1];
    expect(queuedOrder).toBeGreaterThan(addOrder);
  });
});