import { supabase } from './client'; import { config } from '../config'; import { logger } from '../utils/logger'; export interface TurnitinAccount { id: string; email: string; password?: string | null; turnitin_status: string; turnitin_quota_limit: number | null; turnitin_quota_remaining: number | null; turnitin_quota_message: string | null; turnitin_quota_detected_at: string | null; turnitin_first_submission_at: string | null; turnitin_next_retry_at: string | null; turnitin_lease_owner: string | null; turnitin_lease_until: string | null; turnitin_last_checked_at: string | null; turnitin_last_login_at: string | null; turnitin_last_success_at: string | null; turnitin_last_error: string | null; turnitin_session_storage_path: string | null; turnitin_pool_key: string | null; created_at: string; updated_at: string; } export interface AccountQuotaUpdate { turnitin_status?: string; turnitin_quota_limit?: number | null; turnitin_quota_remaining?: number | null; turnitin_quota_message?: string | null; turnitin_quota_detected_at?: string | null; turnitin_next_retry_at?: string | null; turnitin_last_checked_at?: string | null; turnitin_last_success_at?: string | null; turnitin_last_error?: string | null; turnitin_session_storage_path?: string | null; turnitin_pool_key?: string | null; } export interface AccountPoolState { total: number; available: number; running: number; coolingDown: number; quotaLimited: number; loginFailed: number; disabled: number; } function isLegacyHardLimitMessage(account: { turnitin_quota_message?: string | null; turnitin_last_error?: string | null; }): boolean { const text = `${account.turnitin_quota_message || ''} ${account.turnitin_last_error || ''}`.toLowerCase(); return ( text.includes('target class or assignment is not available') || text.includes('login failed') || text.includes('could not log in') || text.includes('credential is invalid') || text.includes('account no longer exists') || /class.*drop/.test(text) || /class.*not.*found/.test(text) || /assignment.*not.*found/.test(text) || text.includes('permanently limited') || text.includes('reached 4 submissions') || text.includes('4-submission limit') ); } function isClaimableLegacyAccount(account: TurnitinAccount, nowMs = Date.now()): boolean { const status = account.turnitin_status || 'available'; const quotaLimit = account.turnitin_quota_limit; const quotaRemaining = account.turnitin_quota_remaining; if (typeof quotaLimit === 'number' && quotaLimit <= 0) return false; const hasQuota = typeof quotaRemaining === 'number' ? quotaRemaining > 0 : status !== 'quota_limited'; const leaseReady = !account.turnitin_lease_until || new Date(account.turnitin_lease_until).getTime() <= nowMs; const retryReady = !account.turnitin_next_retry_at || new Date(account.turnitin_next_retry_at).getTime() <= nowMs; if (!hasQuota || !leaseReady || !retryReady) return false; if (status === 'available' || status === 'cooling_down' || status === 'running') return true; if (status === 'quota_limited') return typeof quotaRemaining === 'number' && !isLegacyHardLimitMessage(account); return false; } /** * Atomically claim an available Turnitin account via database RPC. * Returns the claimed account row, or null if none are available. */ export async function claimAvailableAccount( workerId: string, poolKey = 'modern_lti', ): Promise { // NEW-BUG-2 FIX: Reduced from 20 to 3. When all accounts are stale/cooldown, // spinning 20 times just wastes DB calls. The manager's wait-for-account loop // handles the retry with proper backoff. const MAX_STALE_RETRIES = 3; for (let attempt = 0; attempt < MAX_STALE_RETRIES; attempt++) { let { data, error } = await supabase.rpc('claim_turnitin_identity', { p_worker_id: workerId, p_pool_key: poolKey, }); if ( error && poolKey === 'modern_lti' && /function|schema cache|p_pool_key|claim_turnitin_identity/i.test(error.message) ) { logger.warn('Pool-aware claim RPC is not migrated yet; falling back to legacy modern claim RPC', { error: error.message, }); const fallback = await supabase.rpc('claim_turnitin_identity', { p_worker_id: workerId, }); data = fallback.data; error = fallback.error; } if (error) { logger.error('Failed to claim available account', { poolKey, error: error.message }); throw error; } if (!data || (Array.isArray(data) && data.length === 0)) { return null; } const account = Array.isArray(data) ? data[0] : data; const nextRetryAt = account.turnitin_next_retry_at ? new Date(account.turnitin_next_retry_at).getTime() : null; const quotaRemaining = account.turnitin_quota_remaining; const quotaLimit = account.turnitin_quota_limit; const staleAvailable = (typeof quotaRemaining === 'number' && quotaRemaining <= 0) || (typeof quotaLimit === 'number' && quotaLimit <= 0) || (nextRetryAt !== null && nextRetryAt > Date.now()); if (!staleAvailable) { return account; } const exhaustedQuota = (typeof quotaRemaining === 'number' && quotaRemaining <= 0) || (typeof quotaLimit === 'number' && quotaLimit <= 0); const nextStatus = exhaustedQuota ? 'quota_limited' : 'cooling_down'; const message = exhaustedQuota ? 'Account has no remaining Turnitin quota and was removed from rotation.' : 'Account quota is not currently available.'; logger.warn('Claimed stale unavailable account; removing from immediate rotation', { accountId: account.id, poolKey, quotaLimit, quotaRemaining, nextRetryAt: account.turnitin_next_retry_at, nextStatus, }); await updateAccountQuota(account.id, { turnitin_status: nextStatus, turnitin_quota_remaining: 0, turnitin_quota_message: message, turnitin_next_retry_at: exhaustedQuota ? null : account.turnitin_next_retry_at || new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString(), turnitin_last_error: 'Skipped stale account during claim.', }); await releaseAccount(account.id, nextStatus, 'Skipped stale account during claim.'); } return null; } /** * Claim the exact account that already owns a submitted job. * This is used only for post-submit retries where the worker must reopen the * same Turnitin report viewer instead of submitting the file again. A * quota_limited account is still valid here because report access does not * consume another submission; explicitly disabled/login_failed rows remain blocked. */ export async function claimSpecificAccountForResume( identityId: string, workerId: string, ): Promise { const now = new Date().toISOString(); const leaseUntil = new Date(Date.now() + config.leaseMinutes * 60 * 1000).toISOString(); const { data, error } = await supabase .from('generated_identities') .update({ turnitin_status: 'running', turnitin_lease_owner: workerId, turnitin_lease_until: leaseUntil, updated_at: now, }) .eq('id', identityId) .not('turnitin_status', 'in', '(disabled,login_failed)') .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now},turnitin_lease_owner.eq.${workerId}`) .select('*') .maybeSingle(); if (error) { logger.error('Failed to claim specific resume account', { identityId, error: error.message, }); throw error; } return (data as TurnitinAccount | null) || null; } /** * Lease a specific account for maintenance/quota checks. Returns false when * another worker already owns the lease or the account is not claimable. */ export async function claimSpecificAccount( identityId: string, workerId: string, ): Promise { const now = new Date().toISOString(); const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); const leaseUntil = new Date(Date.now() + config.leaseMinutes * 60 * 1000).toISOString(); const { data, error } = await supabase .from('generated_identities') .update({ turnitin_status: 'running', turnitin_lease_owner: workerId, turnitin_lease_until: leaseUntil, turnitin_last_checked_at: now, updated_at: now, }) .eq('id', identityId) .in('turnitin_status', ['available', 'cooling_down']) .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now}`) .or( [ `and(turnitin_status.eq.available,or(turnitin_quota_remaining.is.null,turnitin_quota_remaining.gt.0),or(turnitin_last_checked_at.is.null,turnitin_last_checked_at.lt.${thirtyMinAgo}))`, `and(turnitin_status.eq.cooling_down,turnitin_next_retry_at.lte.${now})`, ].join(','), ) .select('id') .maybeSingle(); if (error) { logger.error('Failed to claim specific account', { identityId, error: error.message, }); throw error; } return Boolean(data); } /** * Read legacy accounts that were already cooling down before the permanent-limit * cleanup existed. Only accounts whose cooldown window has elapsed are returned, * so we do not interfere with an active post-submit retry. */ export async function getLegacyCooldownAccountsForClassDrop( limit = 2, ): Promise { const now = new Date().toISOString(); const { data, error } = await supabase .from('generated_identities') .select('*') .eq('turnitin_pool_key', 'legacy_carta') .eq('turnitin_status', 'cooling_down') .or(`turnitin_next_retry_at.is.null,turnitin_next_retry_at.lte.${now}`) .order('turnitin_next_retry_at', { ascending: true, nullsFirst: true }) .limit(limit); if (error) { logger.error('Failed to read legacy cooldown accounts for class cleanup', { error: error.message, }); throw error; } return (data as TurnitinAccount[]) || []; } /** * Lease one overdue legacy cooldown account for permanent cleanup. This is * intentionally stricter than generic resume claiming and only targets * legacy_carta/cooling_down rows whose cooldown has elapsed. */ export async function claimLegacyCooldownAccountForClassDrop( identityId: string, workerId: string, ): Promise { const now = new Date().toISOString(); const leaseUntil = new Date(Date.now() + config.leaseMinutes * 60 * 1000).toISOString(); const { data, error } = await supabase .from('generated_identities') .update({ turnitin_status: 'running', turnitin_lease_owner: workerId, turnitin_lease_until: leaseUntil, turnitin_last_checked_at: now, updated_at: now, }) .eq('id', identityId) .eq('turnitin_pool_key', 'legacy_carta') .eq('turnitin_status', 'cooling_down') .or(`turnitin_next_retry_at.is.null,turnitin_next_retry_at.lte.${now}`) .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now}`) .select('*') .maybeSingle(); if (error) { logger.error('Failed to claim legacy cooldown account for class cleanup', { identityId, error: error.message, }); throw error; } return (data as TurnitinAccount | null) || null; } /** * Read one Turnitin account by id. Used by admin maintenance routes before * attempting a manual lease so the previous status can be restored on failure. */ export async function getTurnitinAccountById( identityId: string, ): Promise { const { data, error } = await supabase .from('generated_identities') .select('*') .eq('id', identityId) .maybeSingle(); if (error) { logger.error('Failed to read Turnitin account', { identityId, error: error.message, }); throw error; } return (data as TurnitinAccount | null) || null; } /** * Lease a specific account for an admin-triggered manual quota check. * * This intentionally bypasses the "checked in the last 30 minutes" guard used by * scheduled quota checks, but it still refuses disabled/login-failed accounts and * any account with an active lease. */ export async function claimSpecificAccountForManualQuota( identityId: string, workerId: string, ): Promise { const now = new Date().toISOString(); const leaseUntil = new Date(Date.now() + config.leaseMinutes * 60 * 1000).toISOString(); const { data, error } = await supabase .from('generated_identities') .update({ turnitin_status: 'running', turnitin_lease_owner: workerId, turnitin_lease_until: leaseUntil, turnitin_last_checked_at: now, updated_at: now, }) .eq('id', identityId) .not('turnitin_status', 'in', '(disabled,login_failed)') .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now}`) .select('*') .maybeSingle(); if (error) { logger.error('Failed to claim account for manual quota check', { identityId, error: error.message, }); throw error; } return (data as TurnitinAccount | null) || null; } /** * Count accounts that are immediately claimable. Used to avoid hot-looping a * job when every account is in cooldown/quota_limited state. */ export async function countAvailableAccounts(poolKey?: string): Promise { const now = new Date().toISOString(); if (poolKey === 'legacy_carta') { const { data, error } = await supabase .from('generated_identities') .select('*') .eq('turnitin_pool_key', poolKey) .in('turnitin_status', ['available', 'cooling_down', 'running', 'quota_limited']) .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .or('turnitin_quota_remaining.is.null,turnitin_quota_remaining.gt.0') .or(`turnitin_next_retry_at.is.null,turnitin_next_retry_at.lte.${now}`) .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now}`); if (error) { logger.error('Failed to count legacy available accounts', { poolKey, error: error.message }); throw error; } const nowMs = Date.now(); return ((data as TurnitinAccount[]) || []).filter((account) => isClaimableLegacyAccount(account, nowMs), ).length; } let query = supabase .from('generated_identities') .select('id', { count: 'exact', head: true }) .eq('turnitin_status', 'available') .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .or('turnitin_quota_remaining.is.null,turnitin_quota_remaining.gt.0') .or(`turnitin_next_retry_at.is.null,turnitin_next_retry_at.lte.${now}`) .or(`turnitin_lease_until.is.null,turnitin_lease_until.lte.${now}`); if (poolKey) { query = query.eq('turnitin_pool_key', poolKey); } const { count, error } = await query; if (error) { if ( poolKey === 'modern_lti' && /turnitin_pool_key|column/i.test(error.message) ) { logger.warn('turnitin_pool_key column is not migrated yet; counting all modern accounts without pool filter'); return countAvailableAccounts(); } logger.error('Failed to count available accounts', { poolKey, error: error.message }); throw error; } return count || 0; } /** * Read a compact account pool snapshot for scheduling decisions. The * `available` value is the claimable count, not merely rows with status * available, so stale zero-quota accounts do not keep jobs looping. */ export async function getAccountPoolState(poolKey?: string): Promise { let rowsQuery = supabase .from('generated_identities') .select('turnitin_status'); if (poolKey) { rowsQuery = rowsQuery.eq('turnitin_pool_key', poolKey); } const [available, { data, error }] = await Promise.all([ countAvailableAccounts(poolKey), rowsQuery, ]); if (error) { if ( poolKey === 'modern_lti' && /turnitin_pool_key|column/i.test(error.message) ) { logger.warn('turnitin_pool_key column is not migrated yet; reading global account pool state'); return getAccountPoolState(); } logger.error('Failed to read account pool state', { poolKey, error: error.message }); throw error; } const rows = (data || []) as Array<{ turnitin_status: string | null }>; const counts: AccountPoolState = { total: rows.length, available, running: 0, coolingDown: 0, quotaLimited: 0, loginFailed: 0, disabled: 0, }; for (const row of rows) { const status = row.turnitin_status || 'available'; if (status === 'running') counts.running++; else if (status === 'cooling_down') counts.coolingDown++; else if (status === 'quota_limited') counts.quotaLimited++; else if (status === 'login_failed') counts.loginFailed++; else if (status === 'disabled') counts.disabled++; } return counts; } /** * Release a claimed account back to the pool via database RPC. * Sets the next status and optionally records an error message. */ export async function releaseAccount( identityId: string, nextStatus: string, errorMessage?: string, ): Promise { const { error } = await supabase.rpc('release_turnitin_identity', { p_identity_id: identityId, p_next_status: nextStatus, p_error_message: errorMessage || null, }); if (error) { logger.error('Failed to release account', { identityId, error: error.message }); throw error; } } /** * Update quota-related fields on a Turnitin account. */ export async function updateAccountQuota( identityId: string, update: Partial, ): Promise { const { error } = await supabase .from('generated_identities') .update({ ...update, updated_at: new Date().toISOString() }) .eq('id', identityId); if (error) { logger.error('Failed to update account quota', { identityId, error: error.message }); throw error; } } /** * Get accounts that are due for a quota check: * - Available accounts not checked in the last 30 minutes * - Cooling-down accounts whose next_retry_at has passed */ export async function getAccountsForQuotaCheck( order: 'oldest' | 'newest' = 'oldest', poolKey = 'modern_lti', ): Promise { const thirtyMinAgo = new Date(Date.now() - 30 * 60 * 1000).toISOString(); const now = new Date().toISOString(); // Available accounts not checked recently let availableQuery = supabase .from('generated_identities') .select('*') .eq('turnitin_status', 'available') .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .or('turnitin_quota_remaining.is.null,turnitin_quota_remaining.gt.0') .or(`turnitin_last_checked_at.is.null,turnitin_last_checked_at.lt.${thirtyMinAgo}`); if (poolKey) { availableQuery = availableQuery.eq('turnitin_pool_key', poolKey); } let { data: available, error: err1 } = await availableQuery; if ( err1 && poolKey === 'modern_lti' && /turnitin_pool_key|column/i.test(err1.message) ) { logger.warn('turnitin_pool_key column is not migrated yet; quota check will use global modern account query'); const fallback = await supabase .from('generated_identities') .select('*') .eq('turnitin_status', 'available') .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .or('turnitin_quota_remaining.is.null,turnitin_quota_remaining.gt.0') .or(`turnitin_last_checked_at.is.null,turnitin_last_checked_at.lt.${thirtyMinAgo}`); available = fallback.data; err1 = fallback.error; } if (err1) { logger.error('Failed to get available accounts for quota check', { error: err1.message }); throw err1; } // Cooling-down accounts ready to retry let coolingDownQuery = supabase .from('generated_identities') .select('*') .eq('turnitin_status', 'cooling_down') .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .lte('turnitin_next_retry_at', now); if (poolKey) { coolingDownQuery = coolingDownQuery.eq('turnitin_pool_key', poolKey); } let { data: coolingDown, error: err2 } = await coolingDownQuery; if ( err2 && poolKey === 'modern_lti' && /turnitin_pool_key|column/i.test(err2.message) ) { const fallback = await supabase .from('generated_identities') .select('*') .eq('turnitin_status', 'cooling_down') .or('turnitin_quota_limit.is.null,turnitin_quota_limit.gt.0') .lte('turnitin_next_retry_at', now); coolingDown = fallback.data; err2 = fallback.error; } if (err2) { logger.error('Failed to get cooling-down accounts for quota check', { error: err2.message }); throw err2; } const accounts = [ ...((available as TurnitinAccount[]) || []), ...((coolingDown as TurnitinAccount[]) || []), ]; return accounts.sort((a, b) => { const aChecked = a.turnitin_last_checked_at ? new Date(a.turnitin_last_checked_at).getTime() : null; const bChecked = b.turnitin_last_checked_at ? new Date(b.turnitin_last_checked_at).getTime() : null; if (aChecked === null && bChecked === null) { return new Date(a.created_at).getTime() - new Date(b.created_at).getTime(); } if (order === 'oldest') { if (aChecked === null) return -1; if (bChecked === null) return 1; return aChecked - bChecked; } if (aChecked === null) return 1; if (bChecked === null) return -1; return bChecked - aChecked; }); } /** * Find accounts with expired leases that are still in 'running' status. */ export async function getStaleAccounts(staleSinceMinutes: number): Promise { const cutoff = new Date(Date.now() - staleSinceMinutes * 60 * 1000).toISOString(); const { data, error } = await supabase .from('generated_identities') .select('*') .eq('turnitin_status', 'running') .lt('turnitin_lease_until', cutoff); if (error) { logger.error('Failed to get stale accounts', { error: error.message }); throw error; } return (data as TurnitinAccount[]) || []; } /** * Reset a stale account back to 'available' status. */ export async function resetStaleAccount(identityId: string): Promise { const { error } = await supabase .from('generated_identities') .update({ turnitin_status: 'available', turnitin_lease_owner: null, turnitin_lease_until: null, turnitin_last_error: null, updated_at: new Date().toISOString(), }) .eq('id', identityId); if (error) { logger.error('Failed to reset stale account', { identityId, error: error.message }); throw error; } }