src/db/jobs.ts

import { supabase } from './client';
import { logger } from '../utils/logger';

export interface TurnitinJob {
  id: string;
  user_id: string;
  identity_id: string | null;
  assignment_target_id: string;
  mode: string;
  status: string;
  input_file_path: string;
  input_file_name: string;
  input_file_size: number | null;
  input_file_sha256: string | null;
  submission_request_id: string | null;
  output_pdf_path: string | null;
  output_pdf_expires_at: string | null;
  receipt_pdf_path: string | null;
  receipt_pdf_expires_at: string | null;
  ticket_refunded_at: string | null;
  ticket_refund_reason: string | null;
  viewer_url: string | null;
  similarity_percent: number | null;
  last_completed_step: string | null;
  submission_details: Record<string, unknown> | null;
  filters: Record<string, unknown>;
  attempt_count: number;
  max_attempts: number;
  next_retry_at: string | null;
  error_message: string | null;
  worker_id: string | null;
  started_at: string | null;
  finished_at: string | null;
  created_at: string;
  updated_at: string;
}

/**
 * Atomically claim a pending job using the database RPC.
 * Returns the claimed job row, or null if no jobs are available.
 */
export async function claimPendingJob(workerId: string): Promise<TurnitinJob | null> {
  const { data, error } = await supabase.rpc('claim_turnitin_job', {
    p_worker_id: workerId,
  });

  if (error) {
    logger.error('Failed to claim pending job', { error: error.message });
    throw error;
  }

  if (!data || (Array.isArray(data) && data.length === 0)) {
    return null;
  }

  return Array.isArray(data) ? data[0] : data;
}

/**
 * Update a job's status and optionally merge additional column values.
 */
export async function updateJobStatus(
  jobId: string,
  status: string,
  extra?: Partial<TurnitinJob>,
): Promise<void> {
  const update: Record<string, unknown> = {
    status,
    updated_at: new Date().toISOString(),
    ...extra,
  };

  const { error } = await supabase
    .from('turnitin_jobs')
    .update(update)
    .eq('id', jobId);

  if (error) {
    logger.error('Failed to update job status', { jobId, status, error: error.message });
    throw error;
  }
}

/**
 * Mark a job completed without reviving a job that an administrator or user
 * already moved to a terminal failed/cancelled state while Playwright was
 * finishing in the background.
 */
export async function completeJobIfActive(
  jobId: string,
  fields: Partial<TurnitinJob>,
): Promise<boolean> {
  const update: Record<string, unknown> = {
    ...fields,
    status: 'completed',
    updated_at: new Date().toISOString(),
  };

  const { data, error } = await supabase
    .from('turnitin_jobs')
    .update(update)
    .eq('id', jobId)
    .not('status', 'in', '(failed,cancelled)')
    .select('id')
    .maybeSingle();

  if (error) {
    logger.error('Failed to complete active job', { jobId, error: error.message });
    throw error;
  }

  return Boolean(data);
}

/**
 * Patch arbitrary job fields without changing status.
 * Used for incremental progress checkpoints so retries can resume safely.
 */
export async function updateJobFields(
  jobId: string,
  fields: Partial<TurnitinJob>,
): Promise<void> {
  if (Object.keys(fields).length === 0) return;

  const update: Record<string, unknown> = {
    ...fields,
    updated_at: new Date().toISOString(),
  };

  const { error } = await supabase
    .from('turnitin_jobs')
    .update(update)
    .eq('id', jobId);

  if (error) {
    logger.error('Failed to update job fields', {
      jobId,
      fields: Object.keys(fields),
      error: error.message,
    });
    throw error;
  }
}

/**
 * Fetch a single job by its ID.
 */
export async function getJobById(jobId: string): Promise<TurnitinJob | null> {
  const { data, error } = await supabase
    .from('turnitin_jobs')
    .select('*')
    .eq('id', jobId)
    .single();

  if (error) {
    if (error.code === 'PGRST116') return null; // Row not found
    logger.error('Failed to get job by id', { jobId, error: error.message });
    throw error;
  }

  return data as TurnitinJob;
}

/**
 * Find jobs whose report PDF has expired and should be cleaned up.
 */
export async function getExpiredReportJobs(): Promise<TurnitinJob[]> {
  const now = new Date().toISOString();
  const { data, error } = await supabase
    .from('turnitin_jobs')
    .select('*')
    .or(
      `and(output_pdf_path.not.is.null,output_pdf_expires_at.lte.${now}),` +
      `and(receipt_pdf_path.not.is.null,receipt_pdf_expires_at.lte.${now})`,
    );

  if (error) {
    logger.error('Failed to get expired report jobs', { error: error.message });
    throw error;
  }

  return (data as TurnitinJob[]) || [];
}

/**
 * Atomically increment the attempt_count for a job and return the new value.
 *
 * BUG-1 FIX: The previous Read-Modify-Write approach was a race condition —
 * two Space workers could both read attempt_count=0, compute 1, and both
 * write 1, meaning the limit was never properly enforced.
 *
 * Fix: Use optimistic locking. We UPDATE with a WHERE attempt_count = expected.
 * If the row was modified by another worker concurrently, the update matches
 * 0 rows and we re-read the current value (which already contains the
 * concurrent increment) and return that.
 */
export async function incrementJobAttempt(jobId: string): Promise<number> {
  const job = await getJobById(jobId);
  if (!job) {
    throw new Error(`Job not found: ${jobId}`);
  }

  const expectedCount = (job.attempt_count as number) || 0;
  const newCount = expectedCount + 1;

  const { data, error } = await supabase
    .from('turnitin_jobs')
    .update({ attempt_count: newCount, updated_at: new Date().toISOString() })
    .eq('id', jobId)
    .eq('attempt_count', expectedCount)   // optimistic lock
    .select('attempt_count');

  if (error) {
    logger.error('Failed to increment job attempt', { jobId, error: error.message });
    throw error;
  }

  // If no row was updated another worker already incremented; re-read true value.
  if (!data || data.length === 0) {
    logger.warn('incrementJobAttempt: concurrent update detected; re-reading actual count', { jobId });
    const fresh = await getJobById(jobId);
    return (fresh?.attempt_count as number) || newCount;
  }

  return (data[0] as { attempt_count: number }).attempt_count;
}

/**
 * Find jobs that are stuck in running-like states for longer than the given threshold.
 */
export async function getStaleJobs(staleSinceMinutes: number): Promise<TurnitinJob[]> {
  const cutoff = new Date(Date.now() - staleSinceMinutes * 60 * 1000).toISOString();

  // In our scheme, jobs in running states have started_at or updated_at, we can check updated_at
  const { data, error } = await supabase
    .from('turnitin_jobs')
    .select('*')
    .in('status', [
      'claiming_account',
      'waiting_account',
      'running',
      'uploading',
      'submitted',
      'waiting_similarity',
      'opening_viewer',
      'applying_filters',
      'downloading',
    ])
    .lt('updated_at', cutoff);

  if (error) {
    logger.error('Failed to get stale jobs', { error: error.message });
    throw error;
  }

  return (data as TurnitinJob[]) || [];
}