import crypto from 'node:crypto';
import test from 'node:test';
import assert from 'node:assert/strict';
import { prisma, prismaAdmin } from '@detiq/database';
import { claimNextQueuedJob, claimSpecificJob, requeueStaleRunningJobs } from './job-queue.js';

function id(prefix: string) {
  return `${prefix}_${crypto.randomUUID()}`;
}

async function makeFixtureProject() {
  const tenantId = id('tenant');
  const projectId = id('project');
  await prisma.$transaction(async (tx) => {
    await tx.$executeRawUnsafe(`SELECT set_config('app.current_tenant', $1, true)`, tenantId);
    await tx.tenant.create({ data: { id: tenantId, name: 'Job Queue Test Tenant', slug: tenantId } });
    await tx.project.create({ data: { id: projectId, tenantId, name: 'Job Queue Test Project' } });
  });
  return { tenantId, projectId };
}

async function insertQueuedJob(tenantId: string, projectId: string, type = 'API_TESTS') {
  const jobId = id('job');
  await prismaAdmin.platformJob.create({
    data: { id: jobId, tenantId, projectId, type, status: 'QUEUED', config: { tenantId, projectId } },
  });
  return jobId;
}

test('claimNextQueuedJob: two concurrent claimers never claim the same job (SKIP LOCKED)', async () => {
  const { tenantId, projectId } = await makeFixtureProject();
  const jobIds = await Promise.all([
    insertQueuedJob(tenantId, projectId),
    insertQueuedJob(tenantId, projectId),
    insertQueuedJob(tenantId, projectId),
  ]);

  // Drain the whole queue with concurrent claimers racing each other, in waves, until
  // it's empty — claims globally, so a shared/non-pristine test DB may have other
  // QUEUED jobs lying around ahead of ours in queue order. Draining fully (rather than
  // a fixed attempt count) means this is correct regardless of backlog size, and this
  // test only asserts on the 3 jobs it created, not on global totals.
  const claimedIds: string[] = [];
  for (let wave = 0; wave < 50; wave++) {
    const claims = await Promise.all(Array.from({ length: 6 }, () => claimNextQueuedJob()));
    const got = claims.filter(Boolean).map((c) => c!.id);
    if (got.length === 0) break;
    claimedIds.push(...got);
  }

  // No duplicate claims anywhere (the core SKIP LOCKED guarantee).
  assert.equal(claimedIds.length, new Set(claimedIds).size, 'no job should ever be claimed twice');

  // All 3 of our seeded jobs were claimed exactly once, by some claimer.
  const ourClaims = claimedIds.filter((cid) => jobIds.includes(cid));
  assert.deepEqual([...ourClaims].sort(), [...jobIds].sort());

  const remaining = await prismaAdmin.platformJob.findMany({ where: { id: { in: jobIds } } });
  for (const job of remaining) {
    assert.equal(job.status, 'RUNNING');
    assert.ok(job.workerId, 'claimed job should record a workerId');
    assert.ok(job.heartbeatAt, 'claimed job should get an initial heartbeat');
  }
});

test('claimSpecificJob: returns null for a job that is not QUEUED (already claimed/done)', async () => {
  const { tenantId, projectId } = await makeFixtureProject();
  const jobId = await insertQueuedJob(tenantId, projectId);

  const first = await claimSpecificJob(jobId);
  assert.ok(first, 'first claim should succeed');
  assert.equal(first!.id, jobId);

  const second = await claimSpecificJob(jobId);
  assert.equal(second, null, 'second claim of an already-RUNNING job must be rejected');
});

test('requeueStaleRunningJobs: requeues a RUNNING job whose heartbeat went stale, leaves a fresh one alone', async () => {
  const { tenantId, projectId } = await makeFixtureProject();
  const staleJobId = await insertQueuedJob(tenantId, projectId);
  const freshJobId = await insertQueuedJob(tenantId, projectId);

  const longAgo = new Date(Date.now() - 999 * 60_000); // way past any sane timeout
  const justNow = new Date();
  await prismaAdmin.platformJob.update({
    where: { id: staleJobId },
    data: { status: 'RUNNING', startedAt: longAgo, heartbeatAt: longAgo, workerId: 'dead-worker:1' },
  });
  await prismaAdmin.platformJob.update({
    where: { id: freshJobId },
    data: { status: 'RUNNING', startedAt: justNow, heartbeatAt: justNow, workerId: 'live-worker:1' },
  });

  await requeueStaleRunningJobs();

  const stale = await prismaAdmin.platformJob.findUnique({ where: { id: staleJobId } });
  const fresh = await prismaAdmin.platformJob.findUnique({ where: { id: freshJobId } });

  assert.equal(stale!.status, 'QUEUED', 'stale RUNNING job must be requeued for another worker to pick up');
  assert.equal(stale!.workerId, null, 'requeue must clear the dead worker id');
  assert.equal(fresh!.status, 'RUNNING', 'a job with a recent heartbeat must not be touched, even if RUNNING a while');
});

test('requeueStaleRunningJobs: a job with no heartbeat falls back to startedAt', async () => {
  const { tenantId, projectId } = await makeFixtureProject();
  const jobId = await insertQueuedJob(tenantId, projectId);
  const longAgo = new Date(Date.now() - 999 * 60_000);

  // Simulates a job claimed by the pre-heartbeat code path (or a heartbeat write that
  // never landed) — heartbeatAt is null, only startedAt is set.
  await prismaAdmin.platformJob.update({
    where: { id: jobId },
    data: { status: 'RUNNING', startedAt: longAgo, heartbeatAt: null, workerId: 'dead-worker:2' },
  });

  await requeueStaleRunningJobs();

  const job = await prismaAdmin.platformJob.findUnique({ where: { id: jobId } });
  assert.equal(job!.status, 'QUEUED');
});