diff --git a/package.json b/package.json index a39aa3f..a9ffb55 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "scripts": { "build": "tsx scripts/build.ts", "typecheck": "tsc --noEmit --project tsconfig.json", - "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js", + "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js", "smoke:dist": "node dist/src/cli.js status", "ci": "npm run typecheck && npm test && npm run smoke:dist", "dev": "tsx src/agent.ts", diff --git a/src/supabase-client.ts b/src/supabase-client.ts index 9b047e7..a41d15f 100644 --- a/src/supabase-client.ts +++ b/src/supabase-client.ts @@ -20,6 +20,14 @@ export interface SupabaseMutationOptions { returning?: boolean; } +export interface SupabaseRpcOptions { + /** + * Allows network retries only when the RPC contract itself is idempotent for + * the exact request body. Ordinary database writes must leave this false. + */ + retrySafe?: boolean; +} + export class SupabaseRestError extends Error { constructor( message: string, @@ -85,11 +93,17 @@ function describeCause(error: unknown): string | undefined { return undefined; } -async function fetchSupabase(url: string | URL, init: RequestInit, table: string, operation: string): Promise { - const maxAttempts = 3; +async function fetchSupabase( + url: string | URL, + init: RequestInit, + table: string, + operation: string, + maxAttempts: number +): Promise { + const attempts = Math.max(1, maxAttempts); let lastError: unknown; - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + for (let attempt = 1; attempt <= attempts; attempt += 1) { try { return await fetch(url, { ...init, @@ -97,7 +111,7 @@ async function fetchSupabase(url: string | URL, init: RequestInit, table: string }); } catch (error) { lastError = error; - if (attempt < maxAttempts) { + if (attempt < attempts) { await sleep(500 * attempt); } } @@ -107,7 +121,7 @@ async function fetchSupabase(url: string | URL, init: RequestInit, table: string const causeDetails = describeCause(lastError); const suffix = causeDetails ? ` | cause: ${causeDetails}` : ''; throw new SupabaseNetworkError( - `Supabase ${operation} ${table} network request failed at ${endpoint} after ${maxAttempts} attempts${suffix}`, + `Supabase ${operation} ${table} network request failed at ${endpoint} after ${attempts} attempt${attempts === 1 ? '' : 's'}${suffix}`, table, operation, endpoint, @@ -166,7 +180,7 @@ export async function supabaseSelect( const response = await fetchSupabase(url, { headers: serviceHeaders(), - }, table, 'select'); + }, table, 'select', 3); return parseResponse(response, table); } @@ -183,7 +197,7 @@ export async function supabaseInsert( Prefer: returning ? 'return=representation' : 'return=minimal', }), body: JSON.stringify(body), - }, table, 'insert'); + }, table, 'insert', 1); return parseResponse(response, table); } @@ -205,7 +219,7 @@ export async function supabaseUpsert( ].join(','), }), body: JSON.stringify(body), - }, table, 'upsert'); + }, table, 'upsert', 1); return parseResponse(response, table); } @@ -223,7 +237,7 @@ export async function supabaseUpdate( Prefer: options.returning ? 'return=representation' : 'return=minimal', }), body: JSON.stringify(body), - }, table, 'update'); + }, table, 'update', 1); return parseResponse(response, table); } @@ -238,6 +252,26 @@ export async function supabaseDelete( headers: serviceHeaders({ Prefer: options.returning ? 'return=representation' : 'return=minimal', }), - }, table, 'delete'); + }, table, 'delete', 1); return parseResponse(response, table); } + +export async function supabaseRpc( + functionName: string, + body: Record = {}, + options: SupabaseRpcOptions = {} +): Promise { + if (!/^[a-z0-9_]+$/.test(functionName)) { + throw new Error(`Invalid Supabase RPC function name: ${functionName}`); + } + + const url = `${baseUrl()}/rpc/${functionName}`; + const response = await fetchSupabase(url, { + method: 'POST', + headers: serviceHeaders({ + 'Content-Type': 'application/json', + }), + body: JSON.stringify(body), + }, `rpc/${functionName}`, 'rpc', options.retrySafe ? 3 : 1); + return parseResponse(response, `rpc/${functionName}`); +} diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index bfdff2d..bce06f9 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -8,6 +8,7 @@ import * as instagram from './instagram'; import * as linkedin from './linkedin'; import * as logger from './logger'; import { activePlatformsFromSettings } from './platform-settings'; +import * as workerClaims from './worker-claims'; import * as threads from './threads'; import * as x from './x'; import { buildDailyInventoryPlan, type DailyInventoryQueueRow } from './daily-inventory-planner'; @@ -438,6 +439,13 @@ const OPENAI_SOURCE_BACKOFF_WINDOW_MS = 6 * 60 * 60_000; const OPENAI_GENERATION_BACKOFF_MS = 30 * 60_000; const DEFAULT_OPENAI_TEXT_DAILY_CALL_LIMIT = 40; const DEFAULT_OPENAI_IMAGE_DAILY_CALL_LIMIT = 4; +const SOURCE_CLAIM_LEASE_SECONDS = 300; +const ANGLE_CLAIM_LEASE_SECONDS = 900; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE = 'worker_claims_schema_unavailable'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE = + 'Generation is paused because the database worker-claim contract is not applied.'; +const WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION = + 'Apply the reviewed worker-claims-v1 database migrations before resuming source extraction or drafting.'; interface QueueFromAnglesResult { queued: number; @@ -1501,6 +1509,110 @@ async function assertTenantEntitlement(job: AgentJobRow): Promise { }); } +async function ensureWorkerClaimsReady(job: AgentJobRow, summary: PipelineSummary): Promise { + try { + await workerClaims.assertWorkerClaimsContract(); + return true; + } catch (error) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE); + summary.failedStage ||= 'database_claim_contract'; + summary.failureCode ||= WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE; + if (!summary.errors.includes(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE)) { + summary.errors.push(WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE); + } + await writeWorkerLog(job.user_id, 'warn', WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE, { + jobId: job.id, + kind: job.kind, + expected_contract: workerClaims.WORKER_SCHEMA_CONTRACT, + expected_capabilities: [...workerClaims.REQUIRED_WORKER_CLAIM_CAPABILITIES], + error_type: error instanceof Error ? error.name : 'unknown', + next_action: WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION, + }); + return false; + } +} + +async function releaseSourceClaimBestEffort( + job: AgentJobRow, + record: workerClaims.SourceRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseSourceRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_not_confirmed', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'source_claim_release_uncertain', { + jobId: job.id, + sourceRecordId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function releaseAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const released = await workerClaims.releaseAngleRecordClaim(job.user_id, record); + if (!released) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_release_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + } +} + +async function exhaustAngleClaimBestEffort( + job: AgentJobRow, + record: workerClaims.AngleRecordClaim, + reason: string +): Promise { + try { + const exhausted = await workerClaims.exhaustAngleRecordClaim(job.user_id, record); + if (!exhausted) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_not_confirmed', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + }); + } + return exhausted; + } catch (error) { + await writeWorkerLog(job.user_id, 'warn', 'angle_claim_exhaust_uncertain', { + jobId: job.id, + angleId: record.id, + claimVersion: record.claim_version, + reason, + error: publicError(error), + }); + return false; + } +} + async function loadTenantContext(userId: string): Promise { const settings = (await supabaseSelect('user_settings', { select: '*', @@ -2334,9 +2446,31 @@ async function processBankedSourceRecords( let banked = 0; let queued = 0; - for (const record of records) { + for (const candidate of records) { + if (!isProcessableSourceRecordForAngleExtraction(candidate, sourceUrlsWithAngles)) continue; + + let record: workerClaims.SourceRecordClaim | undefined; + try { + record = (await workerClaims.claimSourceRecordById( + job.user_id, + candidate.id, + workerClaims.createClaimToken(), + SOURCE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + addSummaryError(summary, error); + summary.failedStage ||= 'source_claim'; + summary.failureCode ||= 'source_claim_failed'; + await writeWorkerLog(job.user_id, 'warn', 'source_record_claim_failed', { + jobId: job.id, + sourceRecordId: candidate.id, + error: publicError(error), + }); + break; + } + if (!record) continue; + const sourceText = String(record.source_text || '').trim(); - if (!isProcessableSourceRecordForAngleExtraction(record, sourceUrlsWithAngles)) continue; const fallbackSourceLabel = record.origin === 'authenticated_browser' ? 'browser_collector' : 'manual'; @@ -2362,6 +2496,7 @@ async function processBankedSourceRecords( sourceRecordId: record.id, origin: record.origin || null, source_url_host: safeUrlHost(record.url), + claimVersion: record.claim_version, }); const extractionGuardRequest: OpenAIGenerationGuardRequest = { @@ -2385,6 +2520,7 @@ async function processBankedSourceRecords( summary.failureCode ||= extractionGuard.code; } await writeOpenAIPreflightSkip(job, extractionGuard, extractionGuardRequest); + await releaseSourceClaimBestEffort(job, record, 'generation_preflight_blocked'); break; } @@ -2422,67 +2558,58 @@ async function processBankedSourceRecords( stage: ai.OPENAI_TEXT_ANGLE_EXTRACTION_STAGE, next_action: textError?.nextAction || 'Review the fallback import text, then retry source-record processing.', systemic: textError?.systemic === true, + claimVersion: record.claim_version, }); + await releaseSourceClaimBestEffort(job, record, 'angle_extraction_failed'); if (textError?.systemic) break; continue; } const angles = extraction.angles.slice(0, 5); - if (!angles.length) { - summary.sources.withoutAngles++; - continue; - } - - const angleRows = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ - user_id: job.user_id, - source_record_id: record.id, - source_reddit_post_id: record.reddit_post_id || record.id, - subreddit: record.subreddit || fallbackSourceLabel, - reddit_author: record.reddit_author || fallbackSourceLabel, - source_url: record.url, - angle: `${angle.label}: ${angle.thesis}`, + const angleRows: workerClaims.SourceAngleCommitInput[] = angles.flatMap(angle => tenant.activePlatforms.map(platform => ({ + angle: angle.label + ': ' + angle.thesis, angle_title: angle.label, angle_summary: angle.thesis, intended_platform: platform, - status: 'unused', priority: angle.strength || null, topic: extraction.summary.topic || record.title || null, - used_count: 0, - last_used_at: null, }))); + let committed: workerClaims.SourceAngleCommitResult; try { - await supabaseInsert('angle_records', angleRows); - await supabaseUpdate('source_records', { - used: true, - status: 'exhausted', - }, { - filters: [ - { column: 'id', operator: 'eq', value: record.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); + committed = await workerClaims.commitSourceAngleExtraction(job.user_id, record, angleRows); } catch (error) { addSummaryError(summary, error); - await writeWorkerLog(job.user_id, 'warn', 'source_record_angle_insert_failed', { + summary.failedStage ||= 'source_angle_commit'; + summary.failureCode ||= 'source_claim_commit_failed'; + await writeWorkerLog(job.user_id, 'error', 'source_record_angle_commit_failed', { jobId: job.id, sourceRecordId: record.id, - origin: record.origin || null, + claimVersion: record.claim_version, + requestedAngleCount: angleRows.length, error: publicError(error), + next_action: 'Do not repeat paid extraction automatically. Reconcile the source claim and durable angle rows first.', }); - continue; + break; } - banked += angleRows.length; - summary.angles.created += angleRows.length; + const committedCount = Math.max(0, Number(committed.total_count || 0)); + banked += committedCount; + summary.angles.created += committedCount; + summary.angles.alreadyExisting += Math.max(0, angleRows.length - committedCount); + if (!committedCount) summary.sources.withoutAngles++; sourceUrlsWithAngles.add(record.url); + await writeWorkerLog(job.user_id, 'info', 'source_record_banked_angles', { jobId: job.id, sourceRecordId: record.id, origin: record.origin || null, - angleCount: angleRows.length, + angleCount: committedCount, + claimVersion: record.claim_version, }); + if (!committedCount) continue; + const queuedFromAngles = await queueFromBankedAngles(job, tenant, occupiedSlots, timeZone, summary, openAIActivityLogsToday); queued += queuedFromAngles.queued; if (queued > 0 || await hasActiveBankedAngles(job.user_id)) break; @@ -2574,6 +2701,13 @@ function finalizePipelineSummary(summary: PipelineSummary): void { return; } + if (summary.failureCode === WORKER_CLAIMS_SCHEMA_UNAVAILABLE_CODE) { + summary.outcome = 'blocked'; + summary.message = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_MESSAGE; + summary.nextAction = WORKER_CLAIMS_SCHEMA_UNAVAILABLE_NEXT_ACTION; + return; + } + if (summary.queue.openSlotsAtStart === 0) { summary.outcome = 'deferred'; summary.message = 'The queue already had active items, so the worker did not create another draft.'; @@ -2950,14 +3084,17 @@ async function queueFromBankedAngles( const platform = anglePlatform(angleRow, tenant); if (!platform || !isDraftableAngle(angleRow)) { const reason = !platform ? 'disabled_or_missing_platform' : 'missing_source_metadata'; - await supabaseUpdate('angle_records', { + const rejectedRows = await supabaseUpdate('angle_records', { status: 'rejected', }, { filters: [ { column: 'id', operator: 'eq', value: angleRow.id }, { column: 'user_id', operator: 'eq', value: job.user_id }, + { column: 'status', operator: 'eq', value: 'unused' }, ], + returning: true, }); + if (!rejectedRows.length) continue; result.rejected++; if (summary) { summary.angles.legacyRejected++; @@ -3004,7 +3141,7 @@ async function queueFromBankedAngles( filters: [ { column: 'id', operator: 'eq', value: angleRow.id }, { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, + { column: 'status', operator: 'eq', value: 'unused' }, ], }); } @@ -3035,9 +3172,7 @@ async function queueFromBankedAngles( if (!imageGuard.allowed && imageGuardRequest) { applyDraftPreflightSkip(summary, imageGuard, platform); await writeOpenAIPreflightSkip(job, imageGuard, imageGuardRequest); - if (platform === 'instagram') { - instagramImageGenerationBlocked = true; - } + instagramImageGenerationBlocked = true; continue; } @@ -3070,28 +3205,43 @@ async function queueFromBankedAngles( } continue; } - if (summary) summary.drafts.attempted++; - const locked = await supabaseUpdate('angle_records', { - status: 'in_progress', - }, { - filters: [ - { column: 'id', operator: 'eq', value: angleRow.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ACTIVE_ANGLE_STATUSES }, - ], - returning: true, - }); - const currentAngle = locked[0]; + let currentAngle: workerClaims.AngleRecordClaim | undefined; + try { + currentAngle = (await workerClaims.claimAngleRecordById( + job.user_id, + angleRow.id, + [platform], + workerClaims.createClaimToken(), + ANGLE_CLAIM_LEASE_SECONDS + )).record; + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failureReasons, 'angle_claim_failed'); + summary.failedStage ||= 'angle_claim'; + summary.failureCode ||= 'angle_claim_failed'; + addSummaryError(summary, error); + } + await writeWorkerLog(job.user_id, 'warn', 'angle_record_claim_failed', { + jobId: job.id, + angleId: angleRow.id, + platform, + error: publicError(error), + }); + return result; + } if (!currentAngle) { if (summary) { summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'angle_lock_not_acquired'); + incrementCounter(summary.drafts.skipReasons, 'angle_claim_not_acquired'); } continue; } + if (summary) summary.drafts.attempted++; - const selectedAngle = toAngleCandidateFromRecord(currentAngle); + const selectedAngle = toAngleCandidateFromRecord(currentAngle as AngleRecordRow); const post: RedditPost = { id: currentAngle.source_reddit_post_id || currentAngle.id, title: currentAngle.topic || selectedAngle.label, @@ -3117,8 +3267,9 @@ async function queueFromBankedAngles( cta_goal: '', }; + let draft: Awaited>; try { - const draft = await ai.draftPlatforms( + draft = await ai.draftPlatforms( post, sourceSummary, selectedAngle, @@ -3137,71 +3288,6 @@ async function queueFromBankedAngles( if (platform === 'instagram' && !cloudinary.isCloudinaryUrl(draft.imageUrl)) { throw new WorkerJobError('instagram_image_not_persisted', 'instagram_image_not_persisted'); } - const rows = toQueueRows( - job.user_id, - slot, - post, - currentAngle.source_url || `banked-angle:${currentAngle.id}`, - selectedAngle, - draft, - [platform], - timeZone, - currentAngle.id - ); - if (!rows.length) { - await supabaseUpdate('angle_records', { status: 'exhausted' }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - if (summary) { - summary.drafts.skipped++; - incrementCounter(summary.drafts.skipReasons, 'no_draft_text_created'); - } - continue; - } - - await supabaseInsert('queue_items', rows); - await supabaseUpdate('angle_records', { - status: 'drafted', - used_count: (currentAngle.used_count || 0) + 1, - last_used_at: nowIso(), - }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); - queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); - result.queued += rows.length; - if (summary) { - summary.queue.created += rows.length; - summary.drafts.created += rows.length; - summary.openaiUsageToday.platformDraftsCreatedToday += rows.length; - for (const row of rows) { - incrementCounter(summary.queue.createdByPlatform, String(row.platform || platform)); - } - } - await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { - jobId: job.id, - slotIndex: slot.slotIndex, - localDate: slot.localDate, - localHour: slot.localHour, - scheduledFor: slot.scheduledFor, - timeZone, - angleId: currentAngle.id, - platforms: rows.map(row => row.platform), - }); - if (refreshQueueJobCapacityReached(result.queued)) { - await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { - jobId: job.id, - queuedRows: result.queued, - reason: 'bounded_rows_per_worker_invocation', - }); - break; - } } catch (error) { result.failures++; const imageError = ai.openAIImageErrorDetails(error); @@ -3212,34 +3298,24 @@ async function queueFromBankedAngles( incrementCounter(summary.drafts.failureReasons, imageError?.code || textError?.code || errorFailureReason(error)); if (imageError) { instagramImageGenerationBlocked = platform === 'instagram'; - if (!summary.errors.includes(imageError.userMessage)) { - summary.errors.push(imageError.userMessage); - } + if (!summary.errors.includes(imageError.userMessage)) summary.errors.push(imageError.userMessage); summary.failedStage ||= imageError.stage; summary.failureCode ||= imageError.code; } else if (textError) { - if (!summary.errors.includes(textError.userMessage)) { - summary.errors.push(textError.userMessage); - } + if (!summary.errors.includes(textError.userMessage)) summary.errors.push(textError.userMessage); summary.failedStage ||= textError.stage; summary.failureCode ||= textError.code; } else { addSummaryError(summary, error); } } - await supabaseUpdate('angle_records', { - status: 'unused', - }, { - filters: [ - { column: 'id', operator: 'eq', value: currentAngle.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); + await releaseAngleClaimBestEffort(job, currentAngle, 'platform_draft_failed'); const draftErrorContext = safeErrorContext(error); await writeWorkerLog(job.user_id, 'warn', 'banked_angle_draft_failed', { jobId: job.id, angleId: currentAngle.id, platform, + claimVersion: currentAngle.claim_version, error: publicError(error), ...(draftErrorContext || {}), }); @@ -3253,15 +3329,89 @@ async function queueFromBankedAngles( }); return result; } - if (refreshQueueJobCapacityReached(result.queued, result.failures)) { - await writeWorkerLog(job.user_id, 'warn', 'refresh_queue_job_yielded', { - jobId: job.id, - queuedRows: result.queued, - failures: result.failures, - reason: 'bounded_after_draft_failure', - }); - break; + if (refreshQueueJobCapacityReached(result.queued, result.failures)) break; + continue; + } + + const draftText = getPlatformDraftText(draft, platform).trim(); + if (!draftText) { + const exhausted = await exhaustAngleClaimBestEffort(job, currentAngle, 'no_draft_text_created'); + if (summary) { + summary.drafts.skipped++; + incrementCounter(summary.drafts.skipReasons, exhausted ? 'no_draft_text_created' : 'angle_claim_exhaust_not_confirmed'); + } + continue; + } + + let queueRow: workerClaims.QueueItemCommitResult; + try { + queueRow = await workerClaims.commitClaimedAngleDraft({ + userId: job.user_id, + angleRecordId: currentAngle.id, + claimToken: currentAngle.claim_token, + claimVersion: currentAngle.claim_version, + platform, + slotIndex: slot.slotIndex, + scheduledFor: slot.scheduledFor, + scheduledLocalDate: slot.localDate, + scheduledTimezone: timeZone, + draftText, + instagramImageUrl: platform === 'instagram' ? draft.imageUrl || null : null, + instagramImagePrompt: platform === 'instagram' ? draft.imagePrompt || null : null, + sourceUrl: currentAngle.source_url || 'banked-angle:' + currentAngle.id, + sourceTitle: post.title, + angle: selectedAngle.thesis, + }); + } catch (error) { + result.failures++; + if (summary) { + summary.drafts.failures++; + incrementCounter(summary.drafts.failuresByPlatform, platform); + incrementCounter(summary.drafts.failureReasons, 'angle_claim_commit_failed'); + summary.failedStage ||= 'angle_draft_commit'; + summary.failureCode ||= 'angle_claim_commit_failed'; + addSummaryError(summary, error); } + await exhaustAngleClaimBestEffort(job, currentAngle, 'angle_claim_commit_failed'); + await writeWorkerLog(job.user_id, 'error', 'angle_claim_commit_failed', { + jobId: job.id, + angleId: currentAngle.id, + platform, + claimVersion: currentAngle.claim_version, + error: publicError(error), + next_action: 'Do not regenerate automatically. Reconcile the queue row and claim outcome first.', + }); + return result; + } + + occupiedSlots.add(platformSlotOccupancyKey(platform, slot.localDate, slot.slotIndex)); + queuedAnglePlatformKeys.add(anglePlatformDraftKey(currentAngle.id, platform)); + result.queued++; + if (summary) { + summary.queue.created++; + summary.drafts.created++; + summary.openaiUsageToday.platformDraftsCreatedToday++; + incrementCounter(summary.queue.createdByPlatform, platform); + } + await writeWorkerLog(job.user_id, 'info', 'queued_banked_angle', { + jobId: job.id, + queueItemId: queueRow.id, + slotIndex: slot.slotIndex, + localDate: slot.localDate, + localHour: slot.localHour, + scheduledFor: slot.scheduledFor, + timeZone, + angleId: currentAngle.id, + claimVersion: currentAngle.claim_version, + platforms: [platform], + }); + if (refreshQueueJobCapacityReached(result.queued)) { + await writeWorkerLog(job.user_id, 'info', 'refresh_queue_job_yielded', { + jobId: job.id, + queuedRows: result.queued, + reason: 'bounded_rows_per_worker_invocation', + }); + break; } } @@ -3294,6 +3444,9 @@ async function handleRefreshQueue(job: AgentJobRow, tenant: TenantContext): Prom incrementCounter(summary.drafts.skipReasons, 'no_enabled_platforms'); return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); } + if (!(await ensureWorkerClaimsReady(job, summary))) { + return finishRefreshResult(job, summary, { fetched: 0, banked: 0, queued: 0 }); + } let queued = 0; @@ -4404,6 +4557,14 @@ async function staleJobLogs(job: AgentJobRow): Promise { async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRow[]): Promise { if (job.kind !== 'refresh_queue') return 0; + try { + await workerClaims.assertWorkerClaimsContract(); + } catch { + // During expand-first rollout, do not let missing claim columns block publishing + // or let legacy stale recovery mutate rows it cannot fence safely. + return 0; + } + let released = 0; for (const angleId of staleAngleIdsFromLogs(logs)) { const rows = await supabaseUpdate('angle_records', { @@ -4413,6 +4574,7 @@ async function releaseStaleRefreshAngleLocks(job: AgentJobRow, logs: WorkerLogRo { column: 'id', operator: 'eq', value: angleId }, { column: 'user_id', operator: 'eq', value: job.user_id }, { column: 'status', operator: 'eq', value: 'in_progress' }, + { column: 'claim_token', operator: 'is', value: null }, ], returning: true, }); diff --git a/src/worker-claims.ts b/src/worker-claims.ts new file mode 100644 index 0000000..dfcbae0 --- /dev/null +++ b/src/worker-claims.ts @@ -0,0 +1,328 @@ +import { randomUUID } from 'node:crypto'; + +import { SupabaseRestError, supabaseRpc } from './supabase-client'; +import type { PlatformKey } from './types'; + +export const WORKER_SCHEMA_CONTRACT = 'worker-claims-v1'; +export const REQUIRED_WORKER_CLAIM_CAPABILITIES = [ + 'source-targeted-claim-v1', + 'angle-targeted-claim-v1', + 'angle-exhaust-fenced-v1', + 'source-angle-atomic-commit-v1', + 'angle-queue-atomic-commit-v1', + 'queue-angle-identity-v1', +] as const; + +export interface WorkerSchemaContract { + contract: string; + migration: string; + capabilities: string[]; +} + +export interface SourceRecordClaim { + id: string; + user_id: string; + url: string; + title?: string | null; + origin?: string | null; + score?: number | null; + used: boolean; + fetched_at: string; + created_at: string; + updated_at: string; + reddit_post_id?: string | null; + subreddit?: string | null; + reddit_author?: string | null; + content_hash?: string | null; + status?: string | null; + source_text?: string | null; + claim_token: string; + claim_version: number; + claim_expires_at: string; +} + +export interface AngleRecordClaim { + id: string; + user_id: string; + angle: string; + topic?: string | null; + used_count?: number | null; + created_at?: string | null; + updated_at?: string | null; + source_record_id?: string | null; + source_reddit_post_id?: string | null; + subreddit?: string | null; + reddit_author?: string | null; + source_url?: string | null; + angle_title?: string | null; + angle_summary?: string | null; + intended_platform?: PlatformKey | null; + status?: string | null; + priority?: number | null; + claim_token: string; + claim_version: number; + claim_expires_at: string; +} + +export interface SourceAngleCommitInput { + angle: string; + angle_title: string; + angle_summary: string; + intended_platform: PlatformKey; + priority?: number | null; + topic?: string | null; +} + +export interface SourceAngleCommitResult { + inserted_count: number; + total_count: number; +} + +export interface QueueItemCommitResult { + id: string; + user_id: string; + slot_index: number; + scheduled_for: string; + scheduled_local_date?: string | null; + scheduled_timezone?: string | null; + platform: PlatformKey; + status: string; + draft_text?: string | null; + instagram_image_url?: string | null; + instagram_image_prompt?: string | null; + source_url?: string | null; + source_title?: string | null; + angle?: string | null; + angle_record_id?: string | null; +} + +export interface CommitClaimedAngleDraftInput { + userId: string; + angleRecordId: string; + claimToken: string; + claimVersion: number; + platform: PlatformKey; + slotIndex: number; + scheduledFor: string; + scheduledLocalDate: string; + scheduledTimezone: string; + draftText: string; + instagramImageUrl?: string | null; + instagramImagePrompt?: string | null; + sourceUrl?: string | null; + sourceTitle?: string | null; + angle?: string | null; +} + +export class WorkerClaimsContractError extends Error { + readonly code = 'worker_claims_schema_unavailable'; + + constructor(message: string, public readonly causeDetails?: string) { + super(message); + } +} + +export function createClaimToken(): string { + return randomUUID(); +} + +export async function getWorkerSchemaContract(): Promise { + return supabaseRpc('get_worker_schema_contract', {}, { retrySafe: true }); +} + +export async function assertWorkerClaimsContract(): Promise { + try { + const contract = await getWorkerSchemaContract(); + if (contract?.contract !== WORKER_SCHEMA_CONTRACT) { + throw new WorkerClaimsContractError( + `Expected ${WORKER_SCHEMA_CONTRACT}, received ${String(contract?.contract || 'missing')}` + ); + } + const capabilities = new Set(contract.capabilities || []); + const missing = REQUIRED_WORKER_CLAIM_CAPABILITIES.filter(capability => !capabilities.has(capability)); + if (missing.length) { + throw new WorkerClaimsContractError( + `${WORKER_SCHEMA_CONTRACT} is missing required capabilities: ${missing.join(', ')}` + ); + } + return contract; + } catch (error) { + if (error instanceof WorkerClaimsContractError) throw error; + const detail = error instanceof SupabaseRestError + ? `HTTP ${error.status}` + : error instanceof Error + ? error.name + : 'unknown'; + throw new WorkerClaimsContractError( + `${WORKER_SCHEMA_CONTRACT} is not available on the configured Supabase project`, + detail + ); + } +} + +export async function claimSourceRecordForExtraction( + userId: string, + claimToken = createClaimToken(), + leaseSeconds = 90 +): Promise<{ claimToken: string; record?: SourceRecordClaim }> { + const rows = await supabaseRpc('claim_source_record_for_extraction', { + p_user_id: userId, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function claimSourceRecordById( + userId: string, + sourceRecordId: string, + claimToken = createClaimToken(), + leaseSeconds = 90 +): Promise<{ claimToken: string; record?: SourceRecordClaim }> { + const rows = await supabaseRpc('claim_source_record_by_id', { + p_user_id: userId, + p_source_record_id: sourceRecordId, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function renewSourceRecordClaim( + userId: string, + record: Pick, + leaseSeconds = 90 +): Promise { + return supabaseRpc('renew_source_record_claim', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); +} + +export async function releaseSourceRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('release_source_record_claim', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }); +} + +export async function commitSourceAngleExtraction( + userId: string, + record: Pick, + angles: SourceAngleCommitInput[] +): Promise { + const rows = await supabaseRpc('commit_source_angle_extraction', { + p_user_id: userId, + p_source_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_angles: angles, + }, { retrySafe: true }); + return rows[0] || { inserted_count: 0, total_count: 0 }; +} + +export async function claimAngleRecordForDraft( + userId: string, + platforms: PlatformKey[], + claimToken = createClaimToken(), + leaseSeconds = 300 +): Promise<{ claimToken: string; record?: AngleRecordClaim }> { + const rows = await supabaseRpc('claim_angle_record_for_draft', { + p_user_id: userId, + p_platforms: platforms, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function claimAngleRecordById( + userId: string, + angleRecordId: string, + platforms: PlatformKey[], + claimToken = createClaimToken(), + leaseSeconds = 300 +): Promise<{ claimToken: string; record?: AngleRecordClaim }> { + const rows = await supabaseRpc('claim_angle_record_by_id', { + p_user_id: userId, + p_angle_record_id: angleRecordId, + p_platforms: platforms, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + return { claimToken, record: rows[0] }; +} + +export async function renewAngleRecordClaim( + userId: string, + record: Pick, + leaseSeconds = 300 +): Promise { + return supabaseRpc('renew_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); +} + +export async function releaseAngleRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('release_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }); +} + +export async function exhaustAngleRecordClaim( + userId: string, + record: Pick +): Promise { + return supabaseRpc('exhaust_angle_record_claim', { + p_user_id: userId, + p_angle_record_id: record.id, + p_claim_token: record.claim_token, + p_claim_version: record.claim_version, + }, { retrySafe: true }); +} + +export async function commitClaimedAngleDraft( + input: CommitClaimedAngleDraftInput +): Promise { + const rows = await supabaseRpc('commit_claimed_angle_draft', { + p_user_id: input.userId, + p_angle_record_id: input.angleRecordId, + p_claim_token: input.claimToken, + p_claim_version: input.claimVersion, + p_platform: input.platform, + p_slot_index: input.slotIndex, + p_scheduled_for: input.scheduledFor, + p_scheduled_local_date: input.scheduledLocalDate, + p_scheduled_timezone: input.scheduledTimezone, + p_draft_text: input.draftText, + p_instagram_image_url: input.instagramImageUrl ?? null, + p_instagram_image_prompt: input.instagramImagePrompt ?? null, + p_source_url: input.sourceUrl ?? null, + p_source_title: input.sourceTitle ?? null, + p_angle: input.angle ?? null, + }, { retrySafe: true }); + + const row = rows[0]; + if (!row) { + throw new Error('commit_claimed_angle_draft returned no queue row'); + } + return row; +} diff --git a/test/source-ssrf.test.ts b/test/source-ssrf.test.ts index d30546c..c0bd918 100644 --- a/test/source-ssrf.test.ts +++ b/test/source-ssrf.test.ts @@ -75,7 +75,7 @@ async function main(): Promise { assert.match(worker, /const PROCESSABLE_SOURCE_RECORD_ORIGINS = \[\s*'manual',\s*'authenticated_browser',\s*\] as const/); assert.match(sourceRecordBody, /operator: 'in', value: \[\.\.\.PROCESSABLE_SOURCE_RECORD_ORIGINS\]/); assert.match(sourceRecordBody, /source_text/); - assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\(record, sourceUrlsWithAngles\)/); + assert.match(sourceRecordBody, /isProcessableSourceRecordForAngleExtraction\((?:record|candidate), sourceUrlsWithAngles\)/); assert.match(sourceRecordBody, /extractSourceBankWithJobTimeout/); assert.match(sourceRecordBody, /queueFromBankedAngles/); assert.match(sourceRecordBody, /source_record_selected_for_angle_extraction/); diff --git a/test/supabase-client-retry.test.ts b/test/supabase-client-retry.test.ts new file mode 100644 index 0000000..9f33572 --- /dev/null +++ b/test/supabase-client-retry.test.ts @@ -0,0 +1,125 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import { + SupabaseNetworkError, + supabaseDelete, + supabaseInsert, + supabaseRpc, + supabaseSelect, + supabaseUpdate, + supabaseUpsert, +} from '../src/supabase-client'; + +async function test(name: string, fn: () => Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + console.error(`not ok - ${name}`); + throw error; + } +} + +async function main(): Promise { + const originalFetch = globalThis.fetch; + const originalConfig = { + url: config.SUPABASE_URL, + role: config.SUPABASE_SERVICE_ROLE_KEY, + encryption: config.CREDENTIAL_ENCRYPTION_KEY, + timeout: config.HTTP_TIMEOUT_MS, + }; + + config.SUPABASE_URL = 'https://example.supabase.co'; + config.SUPABASE_SERVICE_ROLE_KEY = 'test-service-role'; + config.CREDENTIAL_ENCRYPTION_KEY = 'test-encryption-key'; + config.HTTP_TIMEOUT_MS = 5_000; + + try { + await test('ordinary Supabase writes never retry ambiguous network failures', async () => { + const operations: Array<() => Promise> = [ + () => supabaseInsert('queue_items', { id: 'one' }), + () => supabaseUpsert('queue_items', { id: 'one' }, 'id'), + () => supabaseUpdate('queue_items', { status: 'ready' }), + () => supabaseDelete('queue_items'), + ]; + + for (const operation of operations) { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + throw new TypeError('simulated network ambiguity'); + }) as typeof fetch; + + await assert.rejects(operation, SupabaseNetworkError); + assert.equal(calls, 1); + } + }); + + await test('Supabase reads retry transient network failures', async () => { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + if (calls < 3) throw new TypeError('transient read failure'); + return new Response('[]', { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof fetch; + + const rows = await supabaseSelect('queue_items', { limit: 1 }); + assert.deepEqual(rows, []); + assert.equal(calls, 3); + }); + + await test('RPC calls do not retry unless the caller declares the contract retry-safe', async () => { + let calls = 0; + globalThis.fetch = (async () => { + calls++; + throw new TypeError('ambiguous RPC failure'); + }) as typeof fetch; + + await assert.rejects( + () => supabaseRpc('unsafe_write', { operation_id: 'one' }), + SupabaseNetworkError + ); + assert.equal(calls, 1); + }); + + await test('retry-safe RPC retries preserve the exact request identity and body', async () => { + let calls = 0; + const bodies: string[] = []; + globalThis.fetch = (async (_input, init) => { + calls++; + bodies.push(String(init?.body || '')); + if (calls === 1) throw new TypeError('transient RPC failure'); + return new Response('{"ok":true}', { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof fetch; + + const body = { + p_user_id: '11111111-1111-1111-1111-111111111111', + p_claim_token: '22222222-2222-2222-2222-222222222222', + }; + const result = await supabaseRpc<{ ok: boolean }>('claim_source_record_for_extraction', body, { + retrySafe: true, + }); + + assert.deepEqual(result, { ok: true }); + assert.equal(calls, 2); + assert.equal(bodies.length, 2); + assert.equal(bodies[0], JSON.stringify(body)); + assert.equal(bodies[1], JSON.stringify(body)); + }); + } finally { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = originalConfig.url; + config.SUPABASE_SERVICE_ROLE_KEY = originalConfig.role; + config.CREDENTIAL_ENCRYPTION_KEY = originalConfig.encryption; + config.HTTP_TIMEOUT_MS = originalConfig.timeout; + } +} + +void main(); diff --git a/test/worker-claims.test.ts b/test/worker-claims.test.ts new file mode 100644 index 0000000..4092f7b --- /dev/null +++ b/test/worker-claims.test.ts @@ -0,0 +1,203 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import { + REQUIRED_WORKER_CLAIM_CAPABILITIES, + WORKER_SCHEMA_CONTRACT, + WorkerClaimsContractError, + assertWorkerClaimsContract, + claimAngleRecordById, + claimSourceRecordById, + commitClaimedAngleDraft, + commitSourceAngleExtraction, + createClaimToken, + exhaustAngleRecordClaim, +} from '../src/worker-claims'; + +async function test(name: string, fn: () => Promise): Promise { + try { + await fn(); + console.log(`ok - ${name}`); + } catch (error) { + console.error(`not ok - ${name}`); + throw error; + } +} + +async function main(): Promise { + const originalFetch = globalThis.fetch; + const originalConfig = { + url: config.SUPABASE_URL, + role: config.SUPABASE_SERVICE_ROLE_KEY, + encryption: config.CREDENTIAL_ENCRYPTION_KEY, + }; + + config.SUPABASE_URL = 'https://example.supabase.co'; + config.SUPABASE_SERVICE_ROLE_KEY = 'test-service-role'; + config.CREDENTIAL_ENCRYPTION_KEY = 'test-encryption-key'; + + try { + await test('claim tokens are caller-owned UUID identities', async () => { + const token = createClaimToken(); + assert.match(token, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i); + }); + + await test('schema capability probe requires the complete worker-claims-v1 cutover', async () => { + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: WORKER_SCHEMA_CONTRACT, + migration: '20260906162000', + capabilities: [...REQUIRED_WORKER_CLAIM_CAPABILITIES], + }), { status: 200 })) as typeof fetch; + + const contract = await assertWorkerClaimsContract(); + assert.equal(contract.contract, WORKER_SCHEMA_CONTRACT); + + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: WORKER_SCHEMA_CONTRACT, + migration: '20260906154500', + capabilities: ['source-angle-atomic-commit-v1'], + }), { status: 200 })) as typeof fetch; + + await assert.rejects( + () => assertWorkerClaimsContract(), + (error: unknown) => error instanceof WorkerClaimsContractError + && error.code === 'worker_claims_schema_unavailable' + && error.message.includes('source-targeted-claim-v1') + ); + }); + + await test('targeted source claim and commit preserve the exact fencing identity', async () => { + const bodies: Array> = []; + const token = '22222222-2222-4222-8222-222222222222'; + const claimVersion = 7; + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push(body); + if (url.endsWith('/rpc/claim_source_record_by_id')) { + return new Response(JSON.stringify([{ + id: 'source-1', + user_id: 'user-1', + url: 'https://example.com/source', + used: false, + fetched_at: '2026-09-06T00:00:00Z', + created_at: '2026-09-06T00:00:00Z', + updated_at: '2026-09-06T00:00:00Z', + status: 'banked', + claim_token: token, + claim_version: claimVersion, + claim_expires_at: '2026-09-06T00:05:00Z', + }]), { status: 200 }); + } + return new Response(JSON.stringify([{ inserted_count: 1, total_count: 1 }]), { status: 200 }); + }) as typeof fetch; + + const claimed = await claimSourceRecordById('user-1', 'source-1', token, 300); + assert.equal(claimed.record?.claim_token, token); + assert.equal(claimed.record?.claim_version, claimVersion); + assert.ok(claimed.record); + + await commitSourceAngleExtraction('user-1', claimed.record, [{ + angle: 'Label: thesis', + angle_title: 'Label', + angle_summary: 'thesis', + intended_platform: 'x', + }]); + + assert.equal(bodies[0].p_source_record_id, 'source-1'); + assert.equal(bodies[0].p_claim_token, token); + assert.equal(bodies[1].p_claim_token, token); + assert.equal(bodies[1].p_claim_version, claimVersion); + assert.equal(bodies[1].p_source_record_id, 'source-1'); + }); + + await test('targeted angle claim and queue commit preserve fencing and schedule identity', async () => { + const bodies: Array> = []; + const token = '33333333-3333-4333-8333-333333333333'; + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push(body); + if (url.endsWith('/rpc/claim_angle_record_by_id')) { + return new Response(JSON.stringify([{ + id: 'angle-1', + user_id: 'user-1', + angle: 'Label: thesis', + source_record_id: 'source-1', + source_reddit_post_id: 'source-1', + subreddit: 'manual', + reddit_author: 'manual', + source_url: 'https://example.com/source', + angle_title: 'Label', + angle_summary: 'thesis', + intended_platform: 'x', + status: 'in_progress', + claim_token: token, + claim_version: 9, + claim_expires_at: '2026-09-06T00:05:00Z', + }]), { status: 200 }); + } + return new Response(JSON.stringify([{ + id: 'queue-1', + user_id: 'user-1', + slot_index: 2, + scheduled_for: '2026-09-07T12:00:00Z', + scheduled_local_date: '2026-09-07', + scheduled_timezone: 'Europe/London', + platform: 'x', + status: 'ready', + draft_text: 'draft', + angle_record_id: 'angle-1', + }]), { status: 200 }); + }) as typeof fetch; + + const claimed = await claimAngleRecordById('user-1', 'angle-1', ['x'], token, 900); + assert.ok(claimed.record); + const row = await commitClaimedAngleDraft({ + userId: 'user-1', + angleRecordId: claimed.record.id, + claimToken: claimed.record.claim_token, + claimVersion: claimed.record.claim_version, + platform: 'x', + slotIndex: 2, + scheduledFor: '2026-09-07T12:00:00Z', + scheduledLocalDate: '2026-09-07', + scheduledTimezone: 'Europe/London', + draftText: 'draft', + }); + + assert.equal(row.id, 'queue-1'); + assert.equal(bodies[0].p_angle_record_id, 'angle-1'); + assert.deepEqual(bodies[0].p_platforms, ['x']); + assert.equal(bodies[1].p_claim_token, token); + assert.equal(bodies[1].p_claim_version, 9); + assert.equal(bodies[1].p_scheduled_local_date, '2026-09-07'); + assert.equal(bodies[1].p_scheduled_timezone, 'Europe/London'); + }); + + await test('no-draft terminal operation keeps the original fencing generation', async () => { + let body: Record = {}; + globalThis.fetch = (async (_input, init) => { + body = JSON.parse(String(init?.body || '{}')) as Record; + return new Response('true', { status: 200 }); + }) as typeof fetch; + + const completed = await exhaustAngleRecordClaim('user-1', { + id: 'angle-1', + claim_token: '44444444-4444-4444-8444-444444444444', + claim_version: 11, + }); + + assert.equal(completed, true); + assert.equal(body.p_angle_record_id, 'angle-1'); + assert.equal(body.p_claim_version, 11); + }); + } finally { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = originalConfig.url; + config.SUPABASE_SERVICE_ROLE_KEY = originalConfig.role; + config.CREDENTIAL_ENCRYPTION_KEY = originalConfig.encryption; + } +} + +void main();