src/worker/manager.ts

import { config } from '../config';
import { logger } from '../utils/logger';
import { sleep } from '../utils/retry';
import { createContext } from './browser-pool';
import { getAccountPassword } from '../crypto/password';
import { supabase } from '../db/client';

// DB imports (will be created by subagent)
import { claimPendingJob, completeJobIfActive, updateJobStatus, updateJobFields, incrementJobAttempt, getJobById, TurnitinJob } from '../db/jobs';
import {
  claimAvailableAccount,
  claimSpecificAccountForResume,
  countAvailableAccounts,
  getAccountPoolState,
  getTurnitinAccountById,
  releaseAccount,
  updateAccountQuota,
  type TurnitinAccount,
} from '../db/accounts';
import { insertJobEvent } from '../db/events';
import { downloadInputFile, uploadReceiptPdf, uploadReportPdf, downloadStorageState, uploadStorageState } from '../db/storage';
import { cancelJob, refundFailedJob } from '../db/tickets';
import { MODERN_ONE_POOL_KEY, runTurnitinJob, RunTurnitinJobInput, RunTurnitinJobResult } from '../engine/turnitin';

import * as path from 'path';
import * as fs from 'fs';
import * as os from 'os';

let activeWorkers = 0;
let running = false;

const ACCOUNT_WAIT_POLL_MS = Number(process.env.ACCOUNT_WAIT_POLL_MS || 15000);
const ACCOUNT_WAIT_MAX_MS = Number(process.env.ACCOUNT_WAIT_MAX_MS || 30 * 60 * 1000);
const RESUME_PROTECTED_STEPS = ['submitted', 'similarity', 'viewer', 'filters', 'download', 'receipt'];
const RESUME_ACCOUNT_RETRY_DELAY_MS = Number(process.env.RESUME_ACCOUNT_RETRY_DELAY_MS || 30000);
const DEFAULT_ACCOUNT_POOL_KEY = 'modern_lti';
const LEGACY_ACCOUNT_POOL_KEY = 'legacy_carta';
const LEGACY_ACCOUNT_QUOTA_LIMIT = 4;

function accountQuotaLimitForPool(poolKey: string, account: TurnitinAccount): number {
  if (poolKey === MODERN_ONE_POOL_KEY) return 1;
  if (poolKey === LEGACY_ACCOUNT_POOL_KEY) return LEGACY_ACCOUNT_QUOTA_LIMIT;
  return account.turnitin_quota_limit || 3;
}

// ---------------------------------------------------------------------------
// In-memory sets to prevent double processing and double refund.
// These are per-process so each HF Space worker keeps its own bookkeeping.
// ---------------------------------------------------------------------------

/** Jobs currently being processed by this worker process. */
const processingJobs = new Set<string>();

/**
 * Jobs that have already been refunded by this worker process.
 * NEW-BUG-4 FIX: Entries are auto-deleted after 1 hour to prevent unbounded growth.
 */
const refundedJobs = new Map<string, NodeJS.Timeout>();

function markAsRefunded(jobId: string): void {
  // Clear existing timer if re-refunding (shouldn't happen, but be safe)
  const existing = refundedJobs.get(jobId);
  if (existing) clearTimeout(existing);
  // Auto-remove after 1 hour
  const timer = setTimeout(() => refundedJobs.delete(jobId), 60 * 60 * 1000);
  // Prevent timer from keeping the process alive during shutdown
  if (timer.unref) timer.unref();
  refundedJobs.set(jobId, timer);
}

function wasRefunded(jobId: string): boolean {
  return refundedJobs.has(jobId);
}

async function refundFailedTicketOnce(
  jobId: string,
  reason: string,
  jobLog: ReturnType<typeof logger.child>,
): Promise<boolean> {
  if (wasRefunded(jobId)) {
    jobLog.info('Refund already attempted by this worker process', { jobId });
    return false;
  }

  try {
    const refunded = await refundFailedJob(jobId, reason);
    markAsRefunded(jobId);
    jobLog.info('Failed job refund checked', { jobId, refunded });
    return refunded;
  } catch (error) {
    jobLog.warn('Failed job refund RPC failed', {
      jobId,
      error: error instanceof Error ? compactWorkerMessage(error.message) : String(error),
    });
    return false;
  }
}

type AssignmentTargetConfig = RunTurnitinJobInput['assignmentTarget'];

function normalizeUiVariant(value: unknown): AssignmentTargetConfig['uiVariant'] {
  return value === 'legacy_carta' ? 'legacy_carta' : 'modern_lti';
}

function defaultAssignmentTarget(): AssignmentTargetConfig {
  return {
    targetUrl: config.turnitinTargetUrl,
    classTitle: config.turnitinClassTitle,
    assignmentTitle: config.turnitinAssignmentTitle as string | null,
    assignmentLaunchUrl: null,
    uiVariant: 'modern_lti',
    accountPoolKey: DEFAULT_ACCOUNT_POOL_KEY,
  };
}

async function loadAssignmentTarget(
  assignmentTargetId: string,
  jobLog: ReturnType<typeof logger.child>,
): Promise<AssignmentTargetConfig> {
  const assignmentTarget = defaultAssignmentTarget();
  if (!assignmentTargetId) return assignmentTarget;

  try {
    const { data, error } = await supabase
      .from('turnitin_assignment_targets')
      .select('*')
      .eq('id', assignmentTargetId)
      .single();

    if (error) throw error;
    if (!data) return assignmentTarget;

    const uiVariant = normalizeUiVariant(data.ui_variant);
    assignmentTarget.targetUrl = data.target_url || assignmentTarget.targetUrl;
    assignmentTarget.classTitle = data.class_title || assignmentTarget.classTitle;
    assignmentTarget.assignmentTitle = data.assignment_title || null;
    assignmentTarget.assignmentLaunchUrl = data.assignment_launch_url || null;
    assignmentTarget.uiVariant = uiVariant;
    assignmentTarget.accountPoolKey =
      data.account_pool_key || (uiVariant === 'legacy_carta' ? 'legacy_carta' : DEFAULT_ACCOUNT_POOL_KEY);

    return assignmentTarget;
  } catch (error) {
    jobLog.warn('Failed to fetch assignment target, using defaults', {
      assignmentTargetId,
      error: error instanceof Error ? error.message : String(error),
    });
    return assignmentTarget;
  }
}

// ---------------------------------------------------------------------------
// Job queue with backpressure — prevents thundering herd
// ---------------------------------------------------------------------------

/**
 * Pending account claim resolvers, FIFO.
 * When a job needs an account and none is free, it pushes a resolve callback
 * here. When any job finishes and releases an account, we pop the first waiter
 * to signal it to retry its claim.
 */
const accountWaiters: Array<() => void> = [];

/** Signal one waiting job that an account may now be available. */
function notifyNextWaiter(): void {
  const next = accountWaiters.shift();
  if (next) next();
}

/** Wait for a notification that an account was released. */
function waitForAccountRelease(timeoutMs: number): Promise<'notified' | 'timeout'> {
  return new Promise((resolve) => {
    const timer = setTimeout(() => {
      // Remove ourselves from the queue on timeout
      const idx = accountWaiters.indexOf(onNotify);
      if (idx >= 0) accountWaiters.splice(idx, 1);
      resolve('timeout');
    }, timeoutMs);
    if (timer.unref) timer.unref();

    const onNotify = () => {
      clearTimeout(timer);
      resolve('notified');
    };
    accountWaiters.push(onNotify);
  });
}

/**
 * Start the worker polling loop.
 * Claims jobs from Supabase and processes them with Playwright.
 */
export async function startWorkerManager(): Promise<void> {
  running = true;
  logger.info('Worker manager started', {
    workerId: config.workerId,
    maxWorkers: config.maxWorkers,
  });

  // BUG-2 FIX: Stale job recovery is now handled EXCLUSIVELY by stale-recovery.ts
  // cron job, which runs every 5 minutes. Removing the duplicate inline loop here
  // prevents two Space workers from simultaneously resetting a job that the other
  // worker is still actively processing.

  while (running) {
    try {
      if (activeWorkers >= config.maxWorkers) {
        await sleep(2000);
        continue;
      }

      // Try to claim a pending job
      const job = await claimPendingJob(config.workerId);

      if (!job) {
        // No jobs available, wait before polling again
        const jitter = config.pollIntervalMs + Math.random() * 2000;
        await sleep(jitter);
        continue;
      }

      // ── Guard: skip job if already being processed by this worker ──
      if (processingJobs.has(job.id)) {
        logger.warn('Skipping job already in progress on this worker', {
          jobId: job.id,
        });
        // Release the claim — set it back to pending so a healthy worker can take it.
        await updateJobStatus(job.id, 'pending').catch(() => {});
        await sleep(1000);
        continue;
      }

      logger.info('Claimed job', { jobId: job.id, mode: job.mode, userId: job.user_id });

      // ── BUG-3 FIX: Defensive guard — verify the claimed job actually belongs ──
      // to this worker. If the Supabase RPC `claim_turnitin_job` accidentally
      // returns a job owned by another worker (e.g. one stuck in 'waiting_account'),
      // processing it here would cause double-submit. Re-read the job from DB to
      // ensure worker_id matches before proceeding.
      const freshJob = await getJobById(job.id);
      if (freshJob && freshJob.worker_id && freshJob.worker_id !== config.workerId) {
        logger.warn('Claimed job belongs to a different worker; releasing', {
          jobId: job.id,
          ownWorker: config.workerId,
          actualWorker: freshJob.worker_id,
        });
        await sleep(1000);
        continue;
      }

      // Mark as in-progress
      processingJobs.add(job.id);

      // Spawn async task for this job (non-blocking)
      activeWorkers++;
      processJob(job)
        .catch((err) => {
          logger.error('Unhandled error in job processing', {
            jobId: job.id,
            error: err instanceof Error ? err.message : String(err),
          });
        })
        .finally(() => {
          activeWorkers--;
          processingJobs.delete(job.id);
        });
    } catch (err) {
      logger.error('Worker manager loop error', {
        error: err instanceof Error ? err.message : String(err),
      });
      await sleep(5000);
    }
  }

  logger.info('Worker manager stopped');
}


/**
 * Stop the worker polling loop gracefully.
 */
export function stopWorkerManager(): void {
  running = false;
  logger.info('Worker manager stop requested');
}

/**
 * Get current active worker count.
 */
export function getActiveWorkerCount(): number {
  return activeWorkers;
}

/**
 * Process a single job end-to-end.
 */
async function processJob(job: TurnitinJob): Promise<void> {
  const jobId = job.id;
  const jobLog = logger.child({ job: jobId });
  let identityId: string | null = null;
  let claimedAccount: TurnitinAccount | null = null;
  let initialLastCompletedStep: string | undefined;
  let currentAttemptCount = (job.attempt_count as number) || 0;
  const maxAttempts = (job.max_attempts as number) || 3;

  // NEW-BUG-3 FIX: Declare tmpDir outside try so the finally block can always clean up.
  const tmpDir = path.join(os.tmpdir(), `turnitin-job-${jobId}`);
  const assignmentTargetId = job.assignment_target_id as string;
  let assignmentTarget = defaultAssignmentTarget();
  let accountPoolKey = DEFAULT_ACCOUNT_POOL_KEY;

  try {
    // ── Re-check job status before processing ──
    // Another worker may have already picked this job up or cancelled it.
    const freshJob = await getJobById(jobId);
    if (!freshJob || !['claiming_account', 'pending'].includes(freshJob.status)) {
      jobLog.warn('Job no longer claimable; skipping', {
        currentStatus: freshJob?.status ?? 'not_found',
      });
      return;
    }

    // Increment attempt count
    currentAttemptCount = await incrementJobAttempt(jobId);

    // If a previous attempt already submitted the file, keep retrying with
    // the same Turnitin account. A different account generally cannot access
    // the existing report viewer URL and must not submit the file again.
    initialLastCompletedStep = freshJob.last_completed_step || undefined;
    const resumeNeedsSameAccount = Boolean(
      initialLastCompletedStep &&
        RESUME_PROTECTED_STEPS.includes(initialLastCompletedStep) &&
        freshJob.identity_id,
    );
    assignmentTarget = await loadAssignmentTarget(assignmentTargetId, jobLog);
    accountPoolKey =
      assignmentTarget.accountPoolKey ||
      (assignmentTarget.uiVariant === 'legacy_carta' ? 'legacy_carta' : DEFAULT_ACCOUNT_POOL_KEY);

    // Claim an available Turnitin account
    await updateJobStatus(jobId, 'claiming_account');
    await emitEvent(jobId, null, 'info', 'claiming_account', 'Looking for available Turnitin account', {
      accountPoolKey,
      uiVariant: assignmentTarget.uiVariant,
    });

    // Before submission, every retry may safely rotate to another account.
    // Keeping a pre-submit job attached to an account that has since become
    // quota_limited makes it wait forever even when the pool has free accounts.
    // Post-submit checkpoints remain pinned to the original account to avoid
    // uploading the same file again.
    const account = resumeNeedsSameAccount
      ? await claimSpecificAccountForResume(freshJob.identity_id as string, config.workerId)
      : await claimAccountForJob(job, identityId, jobLog, accountPoolKey);

    if (!account) {
      if (resumeNeedsSameAccount) {
        const previousAccount = await getTurnitinAccountById(
          freshJob.identity_id as string,
        ).catch(() => null);
        if (
          !previousAccount ||
          ['disabled', 'login_failed'].includes(previousAccount.turnitin_status)
        ) {
          throw new Error(
            'The Turnitin account that owns the submitted file is no longer available for report recovery.',
          );
        }

        const message =
          'Previous submission is still locked by its Turnitin account; retrying shortly.';
        await updateJobStatus(jobId, 'waiting_account', {
          error_message: message,
          attempt_count: Math.max(0, currentAttemptCount - 1),
          next_retry_at: new Date(Date.now() + RESUME_ACCOUNT_RETRY_DELAY_MS).toISOString(),
        });
        await emitEvent(jobId, freshJob.identity_id, 'warning', 'waiting_account', message);
      }
      return;
    }

    claimedAccount = account;
    identityId = account.id;
    jobLog.info('Account claimed', { accountId: identityId, email: account.email });

    if (
      accountPoolKey === MODERN_ONE_POOL_KEY &&
      (
        account.turnitin_quota_limit !== 1 ||
        typeof account.turnitin_quota_remaining !== 'number' ||
        account.turnitin_quota_remaining > 1
      )
    ) {
      await updateAccountQuota(identityId, {
        turnitin_quota_limit: 1,
        turnitin_quota_remaining:
          typeof account.turnitin_quota_remaining === 'number'
            ? Math.min(1, account.turnitin_quota_remaining)
            : 1,
      }).catch(() => {});
      account.turnitin_quota_limit = 1;
      account.turnitin_quota_remaining =
        typeof account.turnitin_quota_remaining === 'number'
          ? Math.min(1, account.turnitin_quota_remaining)
          : 1;
    }

    if (
      accountPoolKey === LEGACY_ACCOUNT_POOL_KEY &&
      (
        account.turnitin_quota_limit !== LEGACY_ACCOUNT_QUOTA_LIMIT ||
        typeof account.turnitin_quota_remaining !== 'number' ||
        account.turnitin_quota_remaining > LEGACY_ACCOUNT_QUOTA_LIMIT
      )
    ) {
      await updateAccountQuota(identityId, {
        turnitin_quota_limit: LEGACY_ACCOUNT_QUOTA_LIMIT,
        turnitin_quota_remaining:
          typeof account.turnitin_quota_remaining === 'number'
            ? Math.min(LEGACY_ACCOUNT_QUOTA_LIMIT, account.turnitin_quota_remaining)
            : LEGACY_ACCOUNT_QUOTA_LIMIT,
      }).catch(() => {});
      account.turnitin_quota_limit = LEGACY_ACCOUNT_QUOTA_LIMIT;
      account.turnitin_quota_remaining =
        typeof account.turnitin_quota_remaining === 'number'
          ? Math.min(LEGACY_ACCOUNT_QUOTA_LIMIT, account.turnitin_quota_remaining)
          : LEGACY_ACCOUNT_QUOTA_LIMIT;
    }

    // Update job with account reference
    await updateJobStatus(jobId, 'running', {
      identity_id: identityId,
      started_at: new Date().toISOString(),
    });
    await emitEvent(jobId, identityId, 'info', 'running', `Starting job with account ${account.email}`);

    // Create temp directory for this job
    fs.mkdirSync(tmpDir, { recursive: true });

    // Download input file from Supabase Storage, except for quota_check jobs.
    const localInputPath = job.mode === 'quota_check'
      ? ''
      : path.join(tmpDir, job.input_file_name as string);
    if (job.mode !== 'quota_check') {
      await downloadInputFile(job.input_file_path as string, localInputPath);
      await emitEvent(jobId, identityId, 'info', 'file_downloaded', 'Input file downloaded');
    }

    // Get storage state for session reuse
    let storageStatePath: string | undefined;
    if (account.turnitin_session_storage_path) {
      const storageState = await downloadStorageState(account.turnitin_session_storage_path);
      if (storageState) {
        storageStatePath = path.join(tmpDir, 'storageState.json');
        fs.writeFileSync(storageStatePath, storageState, 'utf-8');
      }
    }

    // Get password
    const password = getAccountPassword();

    // Parse filters from job
    const filters = (job.filters as Record<string, unknown>) || {};

    // ── Determine resume point ──
    // If a previous attempt already uploaded the file, we must NOT upload
    // again.  The engine's `resumeAfterStep` tells it to skip earlier steps.
    const lastCompletedStep = initialLastCompletedStep;
    const effectiveMode =
      lastCompletedStep && ['submitted', 'similarity', 'viewer', 'filters', 'download', 'receipt'].includes(lastCompletedStep)
        ? 'resubmit' as const  // force resubmit because file is already there
        : (job.mode as 'upload' | 'resubmit' | 'quota_check');

    if (lastCompletedStep) {
      jobLog.info('Resuming job from previous step', {
        lastCompletedStep,
        effectiveMode,
      });
    }

    // Prepare job input
    const jobInput: RunTurnitinJobInput = {
      account: {
        id: identityId,
        email: account.email,
        password,
        quotaLimit: accountQuotaLimitForPool(accountPoolKey, account),
        quotaRemaining:
          typeof account.turnitin_quota_remaining === 'number'
            ? account.turnitin_quota_remaining
            : null,
      },
      assignmentTarget,
      inputFilePath: localInputPath,
      inputFileName: job.input_file_name as string,
      inputFileSize: job.input_file_size ?? undefined,
      outputDir: tmpDir,
      mode: effectiveMode,
      filters: {
        excludeBibliography: filters.excludeBibliography as boolean | undefined,
        excludeQuotes: filters.excludeQuotes as boolean | undefined,
        excludeCitations: filters.excludeCitations as boolean | undefined,
        excludeSmallMatches: filters.excludeSmallMatches as boolean | undefined,
        smallMatchMode: filters.smallMatchMode as 'words' | 'percent' | 'off' | null | undefined,
        smallMatchThreshold: filters.smallMatchThreshold as number | null | undefined,
      },
      storageStatePath,
      resumeAfterStep: lastCompletedStep,
      resumeViewerUrl: freshJob.viewer_url,
      attemptCount: currentAttemptCount,
      onEvent: async (event) => {
        await emitEvent(jobId, identityId, event.level, event.step, event.message, event.metadata);
        await persistJobProgressFromEvent(jobId, event);
      },
    };

    // Run the Playwright job
    const result = await runTurnitinJob(jobInput);

    // Handle quota limit
    if (result.quotaLimit) {
      jobLog.warn('Account quota limited', { limit: result.quotaLimit });
      await updateAccountQuota(identityId, {
        turnitin_status: 'quota_limited',
        turnitin_quota_remaining: 0,
        turnitin_quota_message: result.quotaLimit.message,
        turnitin_quota_detected_at: new Date().toISOString(),
        turnitin_next_retry_at:
          accountPoolKey === MODERN_ONE_POOL_KEY
            ? null
            : new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(),
      });
      await releaseAccount(identityId, 'quota_limited', result.quotaLimit.message);

      const availableAccounts = await countAvailableAccounts(accountPoolKey);
      // Retry with a different account only when one is immediately available.
      if (currentAttemptCount < maxAttempts && availableAccounts > 0) {
        await updateJobStatus(jobId, 'pending');
        await safeUpdateLastCompletedStep(jobId, result.lastCompletedStep);
        await emitEvent(jobId, identityId, 'warning', 'quota_limited', 'Account quota limited, retrying with different account');
      } else {
        await updateJobStatus(jobId, 'failed', {
          error_message: result.quotaLimit.message,
          finished_at: new Date().toISOString(),
        });
        const refunded = await refundFailedTicketOnce(
          jobId,
          result.quotaLimit.message,
          jobLog,
        );
        await safeUpdateLastCompletedStep(jobId, result.lastCompletedStep);
        await emitEvent(jobId, identityId, 'error', 'quota_limited', result.quotaLimit.message, {
          refunded,
        });
      }
      return;
    }

    // Upload PDF report if available
    let outputPdfPath: string | undefined;
    let outputPdfExpiresAt: string | undefined;

    if (result.outputPdfPath && fs.existsSync(result.outputPdfPath)) {
      const uploadResult = await uploadReportPdf(
        job.user_id as string,
        jobId,
        result.outputPdfPath
      );
      outputPdfPath = uploadResult.storagePath;
      outputPdfExpiresAt = uploadResult.expiresAt;
      await emitEvent(jobId, identityId, 'info', 'pdf_uploaded', 'PDF report uploaded to storage');
    }

    let receiptPdfPath: string | undefined;
    let receiptPdfExpiresAt: string | undefined;

    if (result.receiptPdfPath && fs.existsSync(result.receiptPdfPath)) {
      const uploadResult = await uploadReceiptPdf(
        job.user_id as string,
        jobId,
        result.receiptPdfPath,
      );
      receiptPdfPath = uploadResult.storagePath;
      receiptPdfExpiresAt = uploadResult.expiresAt;
      await emitEvent(
        jobId,
        identityId,
        'info',
        'receipt_uploaded',
        'Digital Receipt uploaded to storage',
      );
    }

    // Save storage state for session reuse
    try {
      // The engine should have saved the storage state; we read and upload it
      const stateFile = path.join(tmpDir, 'storageState.json');
      if (fs.existsSync(stateFile)) {
        const stateJson = fs.readFileSync(stateFile, 'utf-8');
        const storagePath = await uploadStorageState(identityId, stateJson);
        await updateAccountQuota(identityId, {
          turnitin_session_storage_path: storagePath,
        });
      }
    } catch {
      jobLog.warn('Failed to save storage state');
    }

    // An administrator may fail/cancel a job while Playwright is already in
    // the viewer. Persist completion only if the job is still active so an
    // in-flight callback cannot revive a terminal job.
    const completionPersisted = await completeJobIfActive(jobId, {
      viewer_url: result.viewerUrl,
      similarity_percent: result.similarityPercent,
      output_pdf_path: outputPdfPath,
      output_pdf_expires_at: outputPdfExpiresAt,
      receipt_pdf_path: receiptPdfPath,
      receipt_pdf_expires_at: receiptPdfExpiresAt,
      error_message: null,
      finished_at: new Date().toISOString(),
    });
    if (!completionPersisted) {
      const terminalArtifacts: Partial<TurnitinJob> = {};
      if (outputPdfPath) terminalArtifacts.output_pdf_path = outputPdfPath;
      if (outputPdfExpiresAt) terminalArtifacts.output_pdf_expires_at = outputPdfExpiresAt;
      if (receiptPdfPath) terminalArtifacts.receipt_pdf_path = receiptPdfPath;
      if (receiptPdfExpiresAt) {
        terminalArtifacts.receipt_pdf_expires_at = receiptPdfExpiresAt;
      }
      if (Object.keys(terminalArtifacts).length > 0) {
        await safeUpdateJobFields(jobId, terminalArtifacts);
      }
    }
    if (result.submissionDetails) {
      await safeUpdateJobFields(jobId, {
        submission_details: result.submissionDetails as Record<string, unknown>,
      });
    }
    await safeUpdateLastCompletedStep(jobId, result.lastCompletedStep || 'download');

    // Insert submission record
    try {
      const { supabase } = await import('../db/client');
      await supabase.from('turnitin_submissions').insert({
        job_id: jobId,
        identity_id: identityId,
        assignment_target_id: assignmentTargetId,
        input_file_name: job.input_file_name as string,
        input_file_size: job.input_file_size,
        input_file_sha256: job.input_file_sha256,
        viewer_url: result.viewerUrl,
        similarity_percent: result.similarityPercent,
        submission_details: result.submissionDetails as Record<string, unknown> | undefined,
        filters_applied: job.filters,
        pdf_path: outputPdfPath,
        pdf_expires_at: outputPdfExpiresAt,
        receipt_pdf_path: receiptPdfPath,
        receipt_pdf_expires_at: receiptPdfExpiresAt,
        submitted_at: result.submittedAt,
      });
    } catch (err) {
      jobLog.error('Failed to insert submission record', {
        error: err instanceof Error ? err.message : String(err),
      });
    }

    // Release account. A final-submission warning means this successful submit
    // likely consumed the last available submission, so keep the account out of
    // rotation until the cooldown window passes.
    if (result.permanentLimit) {
      await updateAccountQuota(identityId, {
        turnitin_status: 'quota_limited',
        turnitin_quota_limit: accountQuotaLimitForPool(accountPoolKey, account),
        turnitin_quota_remaining: 0,
        turnitin_quota_message: result.permanentLimit.message,
        turnitin_quota_detected_at: new Date().toISOString(),
        turnitin_next_retry_at: null,
        turnitin_last_success_at: new Date().toISOString(),
      });
      await releaseAccount(identityId, 'quota_limited', result.permanentLimit.message);
    } else if (result.quotaCooldown) {
      await updateAccountQuota(identityId, {
        turnitin_status: 'cooling_down',
        turnitin_quota_remaining: null,
        turnitin_quota_message: result.quotaCooldown.message,
        turnitin_quota_detected_at: new Date().toISOString(),
        turnitin_next_retry_at: result.quotaCooldown.nextRetryAt,
        turnitin_last_success_at: new Date().toISOString(),
      });
      await releaseAccount(identityId, 'cooling_down', result.quotaCooldown.message);
    } else if (result.quotaWarning) {
      await updateAccountQuota(identityId, {
        turnitin_status: 'quota_limited',
        turnitin_quota_remaining: 0,
        turnitin_quota_message: result.quotaWarning,
        turnitin_quota_detected_at: new Date().toISOString(),
        turnitin_next_retry_at: new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(),
      });
      await releaseAccount(identityId, 'quota_limited', result.quotaWarning);
    } else {
      const remainingFromSubmissionCount =
        typeof result.submissionCount === 'number'
          ? Math.max(0, accountQuotaLimitForPool(accountPoolKey, account) - result.submissionCount)
          : null;
      const currentRemaining =
        typeof account.turnitin_quota_remaining === 'number'
          ? account.turnitin_quota_remaining
          : accountQuotaLimitForPool(accountPoolKey, account);
      const nextRemaining =
        remainingFromSubmissionCount ?? Math.max(0, currentRemaining - 1);
      if (assignmentTarget.uiVariant === LEGACY_ACCOUNT_POOL_KEY && nextRemaining <= 0) {
        const cooldownMessage =
          'Legacy Turnitin account reached 4 submissions. Account is permanently limited because the class should be dropped.';
        await updateAccountQuota(identityId, {
          turnitin_status: 'quota_limited',
          turnitin_quota_remaining: 0,
          turnitin_quota_message: cooldownMessage,
          turnitin_quota_detected_at: new Date().toISOString(),
          turnitin_next_retry_at: null,
          turnitin_last_success_at: new Date().toISOString(),
        });
        await releaseAccount(identityId, 'quota_limited', cooldownMessage);
      } else if (accountPoolKey === MODERN_ONE_POOL_KEY && nextRemaining <= 0) {
        const oneUseMessage =
          'Modern one-use account consumed its single allowed submission. Account is permanently limited.';
        await updateAccountQuota(identityId, {
          turnitin_status: 'quota_limited',
          turnitin_quota_limit: 1,
          turnitin_quota_remaining: 0,
          turnitin_quota_message: oneUseMessage,
          turnitin_quota_detected_at: new Date().toISOString(),
          turnitin_next_retry_at: null,
          turnitin_last_success_at: new Date().toISOString(),
        });
        await releaseAccount(identityId, 'quota_limited', oneUseMessage);
      } else {
        await updateAccountQuota(identityId, {
          turnitin_status: 'available',
          turnitin_quota_remaining: nextRemaining,
          turnitin_quota_message: null,
          turnitin_next_retry_at: null,
          turnitin_last_success_at: new Date().toISOString(),
        });
        await releaseAccount(identityId, 'available');
      }
    }

    if (completionPersisted) {
      await emitEvent(jobId, identityId, 'info', 'completed', `Job completed. Similarity: ${result.similarityPercent ?? 'N/A'}%`);
      jobLog.info('Job completed successfully', {
        similarity: result.similarityPercent,
        viewerUrl: result.viewerUrl,
      });
    } else {
      await emitEvent(
        jobId,
        identityId,
        'warning',
        'terminal_status_preserved',
        'Worker cleanup finished after the job was stopped; terminal status preserved.',
      );
      jobLog.warn('Worker finished after job entered a terminal state; completion was not persisted');
    }

    // Cleanup temp directory (moved to finally, see below)
  } catch (err) {
    const errorMessage = err instanceof Error ? err.message : String(err);
    const publicErrorMessage = compactWorkerMessage(errorMessage);
    logger.error('Job failed', { jobId, error: publicErrorMessage });

    // Try to extract lastCompletedStep from the engine result (it's set
    // on the result object even before throwing because the steps record
    // progress incrementally).  We also check the error object itself in
    // case the engine attached the step there.
    const failedResultStep: string | undefined =
      (err as any)?.lastCompletedStep ||
      undefined;
    const failedViewerUrl: string | undefined = (err as any)?.viewerUrl || undefined;
    const failedSimilarityPercent: number | undefined =
      typeof (err as any)?.similarityPercent === 'number'
        ? (err as any).similarityPercent
        : undefined;

    const isQuotaLimit =
      err instanceof Error &&
      (err.name === 'SubmissionQuotaLimitError' ||
        /reached your limit|submission quota limit/i.test(errorMessage));
    const isTargetUnavailable =
      /waiting for locator\('td\.class_name a|waiting for locator\('tr\.assignment-row|Class title|class.*not found|assignment.*not found|Summer Reading 2026/i.test(
        errorMessage,
      );

    // Determine whether the file was already uploaded on THIS attempt.
    // If so, subsequent retries MUST NOT re-upload.
    // We detect this from the job status transitions that went through
    // the emitEvent calls above (e.g. the 'submitted' event).
    let stepToSave: string | null = failedResultStep || null;
    if (!stepToSave) {
      // Fallback: if the error happened after similarity/viewer/download
      // the job status will have been updated through updateJobStatus.
      const latestJob = await getJobById(jobId).catch(() => null);
      stepToSave = latestJob?.last_completed_step || null;
    }

    const failureProgress: Partial<TurnitinJob> = {};
    if (stepToSave) failureProgress.last_completed_step = stepToSave;
    if (failedViewerUrl) failureProgress.viewer_url = failedViewerUrl;
    if (typeof failedSimilarityPercent === 'number') {
      failureProgress.similarity_percent = failedSimilarityPercent;
    }
    if (Object.keys(failureProgress).length > 0) {
      await safeUpdateJobFields(jobId, failureProgress);
    }

    const isPostSubmitResume = Boolean(
      stepToSave && RESUME_PROTECTED_STEPS.includes(stepToSave),
    );
    const availableAccounts =
      isQuotaLimit || isTargetUnavailable
        ? await countAvailableAccounts(accountPoolKey).catch(() => 0)
        : 1;
    const shouldRetry =
      currentAttemptCount < maxAttempts &&
      (!(isQuotaLimit || isTargetUnavailable) || availableAccounts > 0);
    const submissionConsumedThisAttempt = Boolean(
      isPostSubmitResume && !initialLastCompletedStep,
    );
    const knownRemaining =
      typeof claimedAccount?.turnitin_quota_remaining === 'number'
        ? claimedAccount.turnitin_quota_remaining
        : null;
    const remainingAfterConsumedSubmit =
      knownRemaining === null ? null : Math.max(0, knownRemaining - 1);
    const remainingAfterJob = submissionConsumedThisAttempt
      ? remainingAfterConsumedSubmit
      : knownRemaining;
    const legacyNeedsClassCleanup = Boolean(
      accountPoolKey === LEGACY_ACCOUNT_POOL_KEY &&
        isPostSubmitResume &&
        !shouldRetry &&
        remainingAfterJob === 0,
    );

    // Release account if claimed
    if (identityId) {
      const isLoginError = errorMessage.toLowerCase().includes('login');
      try {
        if (isQuotaLimit) {
          await updateAccountQuota(identityId, {
            turnitin_status: 'quota_limited',
            turnitin_quota_remaining: 0,
            turnitin_quota_message: errorMessage,
            turnitin_quota_detected_at: new Date().toISOString(),
            turnitin_next_retry_at: new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(),
            turnitin_last_error: errorMessage,
          });
          await releaseAccount(identityId, 'quota_limited', errorMessage);
        } else if (isTargetUnavailable) {
          await updateAccountQuota(identityId, {
            turnitin_status: 'quota_limited',
            turnitin_quota_remaining: 0,
            turnitin_quota_message:
              'Target class or assignment is not available. This account is treated as limit because the class may have been dropped.',
            turnitin_quota_detected_at: new Date().toISOString(),
            turnitin_next_retry_at: null,
            turnitin_last_error: errorMessage,
          });
          await releaseAccount(identityId, 'quota_limited', errorMessage);
        } else if (isLoginError) {
          const loginLimitMessage =
            'Turnitin login failed. Account removed from rotation because the credential is invalid or the account no longer exists.';
          await updateAccountQuota(identityId, {
            turnitin_status: 'quota_limited',
            turnitin_quota_remaining: 0,
            turnitin_quota_message: loginLimitMessage,
            turnitin_quota_detected_at: new Date().toISOString(),
            turnitin_next_retry_at: null,
            turnitin_last_error: errorMessage,
          }).catch(() => {});
          await releaseAccount(identityId, 'quota_limited', errorMessage);
        } else {
          let nextStatus = 'available';
          if (isPostSubmitResume) {
            if (shouldRetry) {
              nextStatus = 'cooling_down';
            } else if (legacyNeedsClassCleanup) {
              nextStatus = 'cooling_down';
            } else if (remainingAfterJob === 0) {
              nextStatus = 'quota_limited';
            } else if (remainingAfterJob === null) {
              nextStatus = 'cooling_down';
            }
          }

          if (submissionConsumedThisAttempt) {
            await updateAccountQuota(identityId, {
              turnitin_status: nextStatus,
              turnitin_quota_remaining: remainingAfterConsumedSubmit,
              turnitin_quota_message:
                legacyNeedsClassCleanup
                  ? 'Legacy final submission was consumed, but report processing failed. Class cleanup is pending.'
                  : remainingAfterConsumedSubmit === null
                  ? 'Quota needs refresh after a failed post-submit attempt.'
                  : null,
              turnitin_next_retry_at:
                legacyNeedsClassCleanup
                  ? new Date().toISOString()
                  : nextStatus === 'cooling_down' || nextStatus === 'quota_limited'
                  ? new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString()
                  : null,
              turnitin_last_error: errorMessage,
            }).catch(() => {});
          } else if (nextStatus === 'cooling_down') {
            await updateAccountQuota(identityId, {
              turnitin_status: 'cooling_down',
              turnitin_next_retry_at: legacyNeedsClassCleanup
                ? new Date().toISOString()
                : claimedAccount?.turnitin_next_retry_at,
              turnitin_last_error: errorMessage,
              turnitin_quota_message:
                legacyNeedsClassCleanup
                  ? 'Legacy final submission was consumed, but report processing failed. Class cleanup is pending.'
                  : 'Reserved for retry after submitted file reached report viewer.',
            }).catch(() => {});
          } else if (nextStatus === 'quota_limited') {
            await updateAccountQuota(identityId, {
              turnitin_status: 'quota_limited',
              turnitin_quota_remaining: 0,
              turnitin_quota_message:
                'Submitted file consumed the remaining quota before report processing failed.',
              turnitin_next_retry_at:
                accountPoolKey === MODERN_ONE_POOL_KEY
                  ? null
                  : new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(),
              turnitin_last_error: errorMessage,
            }).catch(() => {});
          }
          await releaseAccount(identityId, nextStatus, errorMessage);
        }
      } catch {
        // Ignore release errors
      }
    }

    // A manual failure/cancellation can happen while Playwright is still
    // unwinding. Preserve that terminal decision after account cleanup.
    const latestJob = await getJobById(jobId).catch(() => null);
    if (latestJob && ['failed', 'cancelled'].includes(latestJob.status)) {
      jobLog.info('Job already terminal after worker failure; retry status not changed', {
        status: latestJob.status,
      });
      return;
    }

    // Check if we should retry
    if (shouldRetry) {
      await updateJobStatus(jobId, 'pending', {
        error_message: publicErrorMessage,
      });
      await safeUpdateLastCompletedStep(jobId, stepToSave);
      await emitEvent(jobId, identityId, 'error', 'failed_retry', `Attempt failed, will retry: ${publicErrorMessage}`);
    } else {
      await updateJobStatus(jobId, 'failed', {
        error_message: publicErrorMessage,
        finished_at: new Date().toISOString(),
      });
      const refunded = await refundFailedTicketOnce(jobId, publicErrorMessage, jobLog);
      await safeUpdateLastCompletedStep(jobId, stepToSave);
      await emitEvent(
        jobId,
        identityId,
        'error',
        'failed_final',
        refunded
          ? `Job failed after ${currentAttemptCount} attempt(s): ${publicErrorMessage}. Ticket refunded.`
          : `Job failed after ${currentAttemptCount} attempt(s): ${publicErrorMessage}`,
        { refunded },
      );
    }
  } finally {
    // NEW-BUG-3 FIX: Always clean up temp dir regardless of success or error
    try {
      if (fs.existsSync(tmpDir)) {
        fs.rmSync(tmpDir, { recursive: true, force: true });
      }
    } catch {
      // Ignore cleanup errors
    }

    // Notify the next waiting job that an account may have been released.
    // This is part of the backpressure system — instead of all waiters
    // polling simultaneously (thundering herd), they wait in a FIFO queue.
    notifyNextWaiter();
  }
}

// ---------------------------------------------------------------------------
// Account claiming with bounded wait & single-refund guarantee
// ---------------------------------------------------------------------------

async function claimAccountForJob(
  job: TurnitinJob,
  identityId: string | null,
  jobLog: ReturnType<typeof logger.child>,
  accountPoolKey: string,
): Promise<TurnitinAccount | null> {
  const startedAt = Date.now();
  let lastWaitingEventAt = 0;

  while (true) {
    // ── Re-check that the job is still ours ──
    // CRITICAL: also check worker_id to catch the case where another Space
    // worker already picked this job while we were sleeping.
    const freshJob = await getJobById(job.id).catch(() => null);
    if (!freshJob) {
      jobLog.warn('Job disappeared while waiting for account', { jobId: job.id });
      return null;
    }
    if (['cancelled', 'completed', 'failed'].includes(freshJob.status)) {
      jobLog.info('Job moved to terminal state while waiting for account; stopping', {
        jobId: job.id,
        status: freshJob.status,
      });
      return null;
    }
    // If another worker took ownership of this job, stop immediately.
    if (freshJob.worker_id && freshJob.worker_id !== config.workerId) {
      jobLog.warn('Job was taken by another worker while waiting for account; stopping', {
        jobId: job.id,
        ownedBy: freshJob.worker_id,
        thisWorker: config.workerId,
      });
      return null;
    }

    const account = await claimAvailableAccount(config.workerId, accountPoolKey);
    if (account) {
      // Double-check ownership one final time to prevent double-processing
      // that can happen when both workers exit the wait loop simultaneously.
      const jobAfterClaim = await getJobById(job.id).catch(() => null);
      if (
        !jobAfterClaim ||
        ['cancelled', 'completed', 'failed'].includes(jobAfterClaim.status) ||
        (jobAfterClaim.worker_id && jobAfterClaim.worker_id !== config.workerId)
      ) {
        jobLog.warn('Job ownership lost right after claiming account; releasing account and stopping', {
          jobId: job.id,
          status: jobAfterClaim?.status,
          ownedBy: jobAfterClaim?.worker_id,
        });
        // Release the just-claimed account back to the pool
        await releaseAccount(account.id, 'available').catch(() => {});
        return null;
      }
      return account;
    }

    const poolState = await getAccountPoolState(accountPoolKey);

    // ── Wait only if there are running accounts AND we haven't timed out ──
    if (poolState.running > 0 && Date.now() - startedAt < ACCOUNT_WAIT_MAX_MS) {
      const message = `No free account right now. Waiting for ${poolState.running} running account(s) to finish.`;
      // Keep status as waiting_account — do NOT reset to claiming_account here.
      // Resetting to claiming_account would allow another worker's claim_turnitin_job
      // RPC to pick up this job, causing double processing.
      await updateJobStatus(job.id, 'waiting_account', {
        error_message: message,
      });

      if (Date.now() - lastWaitingEventAt > 60000) {
        lastWaitingEventAt = Date.now();
        await emitEvent(job.id, identityId, 'warning', 'waiting_account', message, {
          poolState,
        });
      }

      // BACKPRESSURE: Instead of fixed-interval polling (sleep 15s), wait
      // in the FIFO queue for a notification from a finishing job. Fall back
      // to the old sleep interval if no notification arrives.
      await waitForAccountRelease(ACCOUNT_WAIT_POLL_MS);
      // Do NOT reset to claiming_account — keep as waiting_account
      continue;
    }

    // ── No accounts available and nothing running (or timed out) ──
    // Cancel the job and refund the ticket once. The database RPC is
    // idempotent, while the local marker only prevents duplicate attempts
    // after a successful cancellation/refund.
    if (wasRefunded(job.id)) {
      jobLog.warn('Refund already issued for this job; skipping duplicate refund', {
        jobId: job.id,
      });
      // Set job to failed terminal state so neither worker picks it up again.
      await updateJobStatus(job.id, 'failed', {
        error_message: 'All Turnitin accounts are out of quota or cooling down. Ticket already refunded.',
        finished_at: new Date().toISOString(),
      });
      return null;
    }

    const cancelMessage =
      poolState.total === 0
        ? 'No Turnitin accounts are configured. Ticket refunded.'
        : 'All Turnitin accounts are out of quota or cooling down. Ticket refunded.';

    jobLog.warn('No claimable Turnitin accounts; cancelling and refunding ticket', {
      poolState,
    });

    try {
      const cancelled = await cancelJob(job.user_id as string, job.id);
      if (cancelled) {
        markAsRefunded(job.id);
      }
      await emitEvent(job.id, identityId, 'error', 'no_account', cancelMessage, {
        poolState,
        refunded: cancelled,
      });
    } catch (error) {
      const message =
        error instanceof Error ? compactWorkerMessage(error.message) : String(error);

      // If the cancel RPC failed because the job is in a wrong state (e.g.
      // already cancelled by the other worker), just mark it failed.
      jobLog.error('Failed to cancel job', {
        userId: job.user_id,
        jobId: job.id,
        error: message,
      });

      await updateJobStatus(job.id, 'failed', {
        error_message: `${cancelMessage} Refund note: ${message}`,
        finished_at: new Date().toISOString(),
      }).catch(() => {});
      const refunded = await refundFailedTicketOnce(
        job.id,
        `${cancelMessage} Fallback after cancel error: ${message}`,
        jobLog,
      );

      await emitEvent(
        job.id,
        identityId,
        'error',
        'no_account',
        refunded
          ? `${cancelMessage} Refund recovered after cancel error.`
          : `${cancelMessage} Refund note: ${message}`,
        { poolState, refunded },
      );
    }

    return null;
  }
}

function compactWorkerMessage(message: string): string {
  const firstLine = message.split('\n').map((line) => line.trim()).find(Boolean) || message;
  if (/locator\.waitFor: Timeout/i.test(firstLine)) {
    const selector = firstLine.match(/locator\('([^']+)'/i)?.[1];
    return selector
      ? `Turnitin page element did not appear in time: ${selector}`
      : 'Turnitin page element did not appear in time.';
  }
  if (/Upload file was selected, but no Upload\/Confirm\/Submit button was found/i.test(firstLine)) {
    return 'File was selected, but Turnitin did not show a usable upload confirmation button.';
  }
  return firstLine.length > 260 ? `${firstLine.slice(0, 257)}...` : firstLine;
}

/**
 * Helper to emit a job event.
 */
async function emitEvent(
  jobId: string,
  identityId: string | null,
  level: 'info' | 'warning' | 'error',
  step: string,
  message: string,
  metadata?: Record<string, unknown>
): Promise<void> {
  try {
    await insertJobEvent({
      job_id: jobId,
      identity_id: identityId,
      level,
      step,
      message,
      metadata: metadata || {},
    });
  } catch (err) {
    logger.error('Failed to insert job event', {
      jobId,
      step,
      error: err instanceof Error ? err.message : String(err),
    });
  }
}

type EngineEvent = {
  level: 'info' | 'warning' | 'error';
  step: string;
  message: string;
  metadata?: Record<string, unknown>;
};

async function persistJobProgressFromEvent(
  jobId: string,
  event: EngineEvent,
): Promise<void> {
  const fields: Partial<TurnitinJob> = {};

  if (event.step === 'submitted' && event.level === 'info') {
    fields.last_completed_step = 'submitted';
  } else if (
    event.step === 'submission_details' &&
    event.level === 'info' &&
    event.metadata
  ) {
    fields.submission_details = event.metadata;
  } else if (
    event.step === 'similarity' &&
    event.level === 'info' &&
    /^Similarity:/i.test(event.message)
  ) {
    fields.last_completed_step = 'similarity';
    if (typeof event.metadata?.similarityPercent === 'number') {
      fields.similarity_percent = event.metadata.similarityPercent;
    }
  } else if (
    event.step === 'similarity' &&
    event.level === 'info' &&
    typeof event.metadata?.similarityPercent === 'number'
  ) {
    fields.similarity_percent = event.metadata.similarityPercent;
  } else if (
    event.step === 'viewer' &&
    event.level === 'info' &&
    typeof event.metadata?.viewerUrl === 'string'
  ) {
    fields.last_completed_step = 'viewer';
    fields.viewer_url = event.metadata.viewerUrl;
  } else if (
    event.step === 'filters' &&
    event.level === 'info' &&
    /Filter validation passed/i.test(event.message)
  ) {
    fields.last_completed_step = 'filters';
  } else if (
    event.step === 'download' &&
    event.level === 'info' &&
    /PDF downloaded successfully/i.test(event.message)
  ) {
    fields.last_completed_step = 'download';
  } else if (
    event.step === 'receipt' &&
    event.level === 'info' &&
    /Digital Receipt downloaded successfully/i.test(event.message)
  ) {
    fields.last_completed_step = 'receipt';
  }

  if (Object.keys(fields).length > 0) {
    await safeUpdateJobFields(jobId, fields);
  }
}

async function safeUpdateJobFields(
  jobId: string,
  fields: Partial<TurnitinJob>,
): Promise<void> {
  try {
    await updateJobFields(jobId, fields);
  } catch (err) {
    logger.warn('safeUpdateJobFields: could not persist job progress', {
      jobId,
      fields: Object.keys(fields),
      error: err instanceof Error ? err.message : String(err),
    });
  }
}

/**
 * Safely persist the last completed step to the job row.
 * Logs a warning on failure (e.g. column not yet migrated or DB unreachable)
 * but never throws — step tracking must not crash the main job flow.
 */
async function safeUpdateLastCompletedStep(
  jobId: string,
  step: string | null | undefined,
): Promise<void> {
  if (!step) return;
  try {
    const { supabase } = await import('../db/client');
    const { error } = await supabase
      .from('turnitin_jobs')
      .update({ last_completed_step: step, updated_at: new Date().toISOString() })
      .eq('id', jobId);
    if (error) {
      logger.warn('safeUpdateLastCompletedStep: could not persist step (column may not exist yet)', {
        jobId,
        step,
        error: error.message,
      });
    }
  } catch (err) {
    logger.warn('safeUpdateLastCompletedStep: unexpected error', {
      jobId,
      step,
      error: err instanceof Error ? err.message : String(err),
    });
  }
}