diff --git a/docs/reliability-repair-progress.md b/docs/reliability-repair-progress.md index d79a1a7..f0a0338 100644 --- a/docs/reliability-repair-progress.md +++ b/docs/reliability-repair-progress.md @@ -1,63 +1,78 @@ # Reliability repair progress -This file records only implementation status that is evidenced by repository changes and CI. It does not claim production deployment or provider readiness. +This file records implementation status, not production deployment or provider readiness. The final exact-SHA CI receipts are recorded on the relevant pull requests. A passing component test is not live account evidence. ## Sequence 1: release identity and complete CI gate -Status: merged to upstream `main` through PR #2. The final pre-merge branch head was proven by the complete upstream pull-request CI gate before merge. +Merged to upstream through PR #2. `npm run ci` gates typecheck, the complete normal regression suite and the compiled-runtime smoke check. Deployment invokes that same gate. Cloudflare version metadata and the Git SHA are exposed separately from provider readiness. -Evidence: +Hosted Threads and Instagram remain unavailable; Facebook remains paused; LinkedIn compatibility is unverified; X requires a tenant-owned connection. No production deployment is claimed here. -- `npm run ci` gates typecheck, the complete repository test suite, and the compiled runtime smoke check. -- the deploy workflow runs that same gate before Wrangler deployment. -- Cloudflare version metadata and the deployment Git SHA are exposed separately from provider readiness. -- hosted Threads and Instagram publication remain explicitly unavailable; Facebook remains paused; LinkedIn compatibility remains unverified; X remains tenant-scoped. -- no production deployment is claimed by this change. +## D03 containment and async runtime isolation -## D03 containment: overlapping tenant runtime state +Upstream PR #3 added a shared per-isolate exclusive gate for scheduled and authenticated tick drains. Its regression proves overlapping drains do not enter mutable tenant runtime concurrently and rejection releases the gate. -Status: merged to upstream `main` through PR #3 after fresh upstream CI passed on the branch synchronised to the PR #2 merge commit. +Upstream PR #4 added async-local configuration and token callback isolation. Its recorded pre-merge head was `f68e4434efc18053d11c85ff1b3d2dd334c5dc3d`; merge commit `033b9b578c120bec0b725311eba1db3d3bfe5530`. -The Cloudflare scheduled and authenticated `/tick` job-drain entry points share one per-isolate exclusive run gate. This prevents two overlapping SaaS drains in the same Worker isolate from entering the mutable tenant runtime concurrently. Separate Worker isolates do not share process globals. +The Worker installs scoped configuration accessors after loading Cloudflare bindings. Each SaaS drain runs inside `AsyncLocalStorage`. Tenant values and Threads/LinkedIn/X token-persistence callbacks stay in the originating execution. Local single-tenant behaviour is retained outside that scope. `processPendingSupabaseJobs()` remains serial and the exclusive drain gate remains in place. -The containment regression deliberately interleaves two executions, proves maximum concurrent execution is one, and proves a rejected execution releases the gate. +This addresses cross-execution process-global leakage. It is not a claim that every provider now uses explicit immutable client arguments, or that credential-version/disconnect fencing exists. Those connection-lifecycle contracts remain separate work. -## D03 runtime isolation: process-global config and token callbacks +## D22: explicit platform activation -Status: merged to upstream `main` through PR #4. Fresh upstream CI run #88 passed `npm ci` and the complete `npm run ci` gate on synchronised head `f68e4434efc18053d11c85ff1b3d2dd334c5dc3d` before merge. The upstream PR #4 merge commit is `033b9b578c120bec0b725311eba1db3d3bfe5530`. +Present in this branch: only exact persisted boolean `true` enables a platform. Missing rows, fields, null and false fail closed. The schema-owner bootstrap also contains default-false settings and matching UI semantics. Existing legacy true values are not mass-rewritten because their original intent cannot be inferred safely. -The Worker installs async-scoped accessors on the existing config object only after Cloudflare bindings have been copied into `process.env`. Each scheduled/authenticated SaaS drain then runs inside its own `AsyncLocalStorage` context. +## D04/D05/D06: source and angle claims and safe transport -Within that context: +This branch is stacked on `codex/atomic-worker-claims`, head `984d0afa1fdfd696153fa8615615b1e2d42596e3`. That dependency introduces the typed worker-claims contract, database-owned source/angle leases and fencing, atomic generation finalisation, and explicit Supabase RPC retry semantics. -- tenant config writes made by the existing `withTenantRuntime` path are copy-on-write and remain inside the current async execution instead of mutating process-global values; -- OpenAI, Cloudinary, Instagram, Facebook and provider modules that already read the shared config object transparently resolve the scoped values without a flag-day call-signature rewrite; -- Threads, LinkedIn and X token-persistence setters use scope-local callback slots when a SaaS runtime scope exists; -- token rotation updates only the current scoped config snapshot while the Supabase persistence callback remains attached to that same async execution; -- outside a SaaS runtime scope, the existing local single-tenant behaviour is preserved. +Ordinary ambiguous mutations are single-attempt. Only reads and RPCs whose exact request identities are designed to be idempotent opt into retries. This does not claim that every agent job/enqueue path already has durable uniqueness or fencing. -The current `processPendingSupabaseJobs()` implementation remains serial, so tenant runtime mutation is restored between jobs inside a drain. The earlier exclusive run gate remains defence-in-depth but is no longer the only boundary preventing overlapping Worker invocations from sharing config or token callbacks. +## D07/D08/D09: connected publication execution -The proven regression covers overlapping tenant credentials, scope-local Threads/LinkedIn/X persistence callbacks, rotated-token isolation, base-config isolation and failure cleanup. Explicit provider-client arguments remain desirable architectural cleanup, but the process-global cross-tenant safety defect is no longer the active blocker. +The hosted Worker consumer is now wired in `codex/publication-ledger-v1`, fork PR `AyobamiH/social-agents#4`, not merely a collection of unused helper modules. -No provider is re-enabled and no deployment was performed by these D03 repairs. +### Execution boundary -## D22: fail-closed tenant platform activation +`publishQueueRow()` delegates to `executePublication()`: -Status: implemented on `codex/fail-closed-platform-settings`; upstream pull-request CI evidence is required before merge. +1. Probe the exact `publication-ledger-v1` schema/capability contract. +2. Read existing outcome by tenant and queue identity. Accepted, rejected, dispatching or unknown publications never become a blind resend. +3. Reject unavailable hosted Meta and paused Facebook before claim, token refresh or paid media work. +4. Claim one immutable queue snapshot with a caller-owned token and fencing version. +5. Prepare credentials, identify the provider account, and recheck entitlement, explicit enablement and scheduled automation/due time. +6. Persist the dispatch attempt before the provider write. +7. Publish only the database snapshot and make one provider publishing request per attempt. +8. Record acceptance and the queue/history projection transactionally through the ledger. -Tenant platform activation now has one explicit policy: a platform is active only when its persisted `*_enabled` setting is exactly boolean `true`. Missing settings rows, missing fields, `null` and `false` all remain disabled. +Provider dispatch and database finalisation have different error boundaries. A database failure after provider success cannot be interpreted as a provider rejection. Lost begin-dispatch responses cause no provider call. Uncertain provider outcomes stay unknown and non-retryable. -The regression covers: +Post-acceptance angle/telemetry failures cannot reopen sending. Where possible the result includes the exact history ID as well as intent, attempt and provider IDs. Provider acceptance and later visibility verification remain separate. -- a missing settings object enabling no platforms; -- all-null flags enabling no platforms; -- all-false flags enabling no platforms; -- mixed settings enabling only explicit `true` entries; -- canonical platform ordering when all five platforms are explicitly enabled. +### Recovery boundary -No provider is re-enabled, no credential semantics change, no queue or billing behaviour changes, and no deployment is performed by this repair. +`stalePublishJobResult()` calls exact ledger reconciliation. The source-URL/platform/time history matcher has been removed from production orchestration. `recoverStalePublications()` also scans orphan claims/dispatches independently of parent job status, including publish-all work. -## Next bounded repair +Expired pre-dispatch ownership may be released with its token/fence. Stale dispatches become unknown. Legacy publishing rows without ledger identity remain quarantined, not relabelled failed so they can be retried. -After D22 is green and merged, proceed to atomic database claims/fencing for jobs, sources and angles, then publication-attempt identity and unknown-outcome handling. Meta publication remains disabled until the publication ledger and provider-specific restoration work are ready. +A delayed positive provider response can resolve its own unknown attempt using exact attempt ID, dispatch operation ID, external object ID and recorded response evidence. Negative search results are not sufficient evidence. Visibility is not inferred from acceptance. + +### Adapter corrections and preserved limits + +X no longer refreshes and repeats POST after an auth-looking response. Read-only identity verification may refresh after explicit HTTP 401. LinkedIn uses its returned object ID or `x-restli-id`, never the placeholder `posted`. + +This does not migrate LinkedIn to the newer Posts API, restore Meta, grant provider scopes, or implement new formats. No provider is newly enabled. Connection rotation/removal version fences and full revision-approval lifecycle still need their own release evidence. The ledger guarantees the claimed snapshot is what is sent; it does not manufacture an earlier approval record. + +### Test boundary and ownership + +The normal regression gate includes executor fault injection, provider single-dispatch checks and delayed receipt reconciliation. `test/publication-database.integration.ts` exercises the actual Worker, encryption, async tenant contexts, Supabase REST/RPC transport and a real local Postgres database. Provider network calls are intercepted: this is integration/failure evidence, not live-provider evidence. + +The cross-repository integration workflow belongs to the private schema-owner repository, `AyobamiH/oneclickpostfactory`, at `.github/workflows/publication-worker-integration.yml`. It checks out an exact public Worker SHA and its own schema. No broad cross-repository secret is needed and no private schema is copied into this public repository. The integration must be rerun with a new exact Worker pin whenever the consumer changes. + +The one-time hash-guarded source-edit workflow and script were removed after committing the source delta. No write-enabled test/codemod workflow remains from this cutover. + +## Production remains gated + +Neither schema nor consumer is deployed by this branch. Before rollout: review the paired PRs, establish canonical schema ownership, apply to isolated staging first, verify the capability receipt, drain old publication dispatch owners, quarantine unresolved legacy publications, then test a bounded authorised cohort. Never run old and new executors as competing owners of the same queue row. Rollback must preserve ledger-owned unknown states and must not replay the legacy queue. + +Remaining programme work includes connection lifecycle/version fences, billing inbox and entitlements, durable generation budgets, fair scheduling and typed UI recovery, provider restoration/compatibility, and account-authorised scheduled canaries. Do not describe the whole SaaS as production-repaired based on this publication slice alone. diff --git a/package.json b/package.json index a9ffb55..7be1961 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 && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.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 && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.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/linkedin.ts b/src/linkedin.ts index f9f1822..618dfa4 100644 --- a/src/linkedin.ts +++ b/src/linkedin.ts @@ -189,7 +189,9 @@ export function publish(text: string): Promise { timeoutMs: config.HTTP_TIMEOUT_MS, }).then(({ status, headers, data, rawText }) => { if (status === 201) { - return data.id || 'posted'; + const id = data.id || headers.get('x-restli-id'); + if (typeof id !== 'string' || !id.trim()) throw new Error('LinkedIn acceptance receipt missing'); + return id; } throw normalizeLinkedInError('post', status, headers.get('content-type'), rawText, data); }).catch(error => { diff --git a/src/publication-executor.ts b/src/publication-executor.ts new file mode 100644 index 0000000..40ed525 --- /dev/null +++ b/src/publication-executor.ts @@ -0,0 +1,379 @@ +import * as ledger from './publication-ledger'; +import { classifyPostDispatchError } from './publication-outcome'; +import { recordObservedAcceptance } from './publication-receipts'; +import { supabaseSelect } from './supabase-client'; +import type { PlatformKey } from './types'; + +export interface PublicationTarget { + userId: string; + queueItemId: string; + platform: PlatformKey; +} + +export interface ProviderAcceptance { + externalPostId: string; +} + +export interface PreparedPublication { + providerAccountRef: string; + // Exactly one provider write. Refresh/identity checks belong in prepare(), not send(). + send(): Promise; +} + +export interface PublicationHooks { + prepare(payload: Readonly): Promise; + afterAccepted?(attempt: ledger.PublicationAttempt, payload: Readonly): Promise; +} + +export type PublicationExecutionResult = Record & { + outcome: 'accepted' | 'verified' | 'unknown' | 'rejected' | 'blocked' | 'retry_wait'; + jobStatus: 'completed' | 'completed_with_errors' | 'failed'; + failureCode: string | null; + summary: Record; +}; + +export class PublicationPreflightError extends Error { + constructor(public readonly code: string) { + super(code); + this.name = 'PublicationPreflightError'; + } +} + +const UNKNOWN_ACTION = 'Do not retry or recreate this post. Reconcile the exact publication attempt before another dispatch.'; +const SAFE_RETRY_ACTION = 'This execution did not dispatch. Reconcile any existing attempt or resolve the blocker before retrying this same queue item.'; + +type Ledger = Pick; + +const productionLedger: Ledger = { + ...ledger, + recordPublicationAccepted: recordObservedAcceptance, +}; + +function result( + target: PublicationTarget, + outcome: PublicationExecutionResult['outcome'], + code: string | null, + state: ledger.PublicationStateSnapshot = {}, + extra: Record = {} +): PublicationExecutionResult { + const accepted = outcome === 'accepted' || outcome === 'verified'; + const message = outcome === 'verified' + ? 'Provider acceptance and later visibility verification are recorded.' + : outcome === 'accepted' + ? 'Provider acceptance is recorded. Visibility has not been verified.' + : outcome === 'unknown' + ? 'Publication outcome requires reconciliation. No automatic resend is allowed.' + : outcome === 'rejected' + ? 'The provider rejected this attempt. No acceptance receipt was created.' + : 'Publication is blocked before a new dispatch.'; + const nextAction = accepted ? 'No resend is needed.' + : outcome === 'unknown' ? UNKNOWN_ACTION + : outcome === 'rejected' ? 'Resolve the provider rejection, then use the authorised recovery flow.' + : SAFE_RETRY_ACTION; + const details = { + outcome, + message, + nextAction, + failureCode: code, + platform: target.platform, + queueItemId: target.queueItemId, + publicationIntentId: state.intent?.id || null, + publicationAttemptId: state.attempt?.id || null, + dispatchOperationId: state.attempt?.dispatch_operation_id || null, + externalPostId: state.attempt?.external_post_id || null, + publishHistoryId: state.history?.id || null, + visibilityVerified: outcome === 'verified', + automaticRetryAllowed: false, + ...extra, + }; + return { + ...details, + outcome, + failureCode: code, + jobStatus: accepted ? 'completed' : 'failed', + summary: { ...details, failedStage: code ? 'publication' : null, errors: code ? [code] : [] }, + }; +} + +function validIntent(target: PublicationTarget, intent: ledger.PublicationIntent): boolean { + return intent.user_id === target.userId + && intent.queue_item_id === target.queueItemId + && intent.platform === target.platform + && intent.payload?.platform === target.platform; +} + +function validAttempt(target: PublicationTarget, state: ledger.PublicationStateSnapshot): boolean { + const { intent, attempt, history } = state; + if (!intent || !validIntent(target, intent)) return false; + if (attempt && (attempt.user_id !== target.userId + || attempt.queue_item_id !== target.queueItemId + || attempt.intent_id !== intent.id + || attempt.platform !== target.platform)) return false; + if (history && (!attempt || history.user_id !== target.userId + || history.queue_item_id !== target.queueItemId + || history.publication_intent_id !== intent.id + || history.publication_attempt_id !== attempt.id + || history.platform !== target.platform + || history.external_post_id !== attempt.external_post_id)) return false; + return true; +} + +function terminalResult(target: PublicationTarget, state: ledger.PublicationStateSnapshot): PublicationExecutionResult | undefined { + if (!state.intent) return undefined; + if (!validAttempt(target, state)) return result(target, 'unknown', 'publication_identity_mismatch'); + // The latest attempt is authoritative if separate read queries straddled an outcome commit. + const status = state.attempt?.state || state.intent.state; + if (status === 'accepted' || status === 'verified') { + if (!state.attempt?.external_post_id) return result(target, 'unknown', 'publication_receipt_incomplete', state); + return result(target, status, null, state); + } + if (status === 'unknown' || status === 'dispatching') { + return result(target, 'unknown', 'publication_outcome_unknown', state); + } + if (status === 'rejected') return result(target, 'rejected', 'publication_rejected', state); + return undefined; +} + +async function readBestEffort(target: PublicationTarget, db: Ledger): Promise { + try { + return await db.loadPublicationStateForQueueItem(target.userId, target.queueItemId); + } catch { + return {}; + } +} + +async function recordUnknownBestEffort( + target: PublicationTarget, + intent: ledger.PublicationIntent, + attempt: ledger.PublicationAttempt, + code: string, + db: Ledger, + observedExternalPostId?: string +): Promise { + try { + const recorded = await db.recordPublicationUnknown({ + userId: target.userId, + attempt, + errorCode: code, + errorMessage: UNKNOWN_ACTION, + providerReceipt: observedExternalPostId ? { observed_external_post_id: observedExternalPostId } : null, + }); + return result(target, 'unknown', code, { intent, attempt: recorded }, { + bookkeepingPending: Boolean(observedExternalPostId), + ...(observedExternalPostId ? { externalPostId: observedExternalPostId, providerAccepted: true } : {}), + }); + } catch { + // Acceptance may have committed despite a lost response, or won a recovery race. + const current = await readBestEffort(target, db); + const terminal = terminalResult(target, current); + if (terminal && (terminal.outcome === 'accepted' || terminal.outcome === 'verified')) return terminal; + return result(target, 'unknown', code, { intent, attempt }, { + bookkeepingPending: true, + ...(observedExternalPostId ? { externalPostId: observedExternalPostId, providerAccepted: true } : {}), + }); + } +} + +/** No external write is possible without a confirmed durable dispatch attempt. */ +export async function executePublication( + target: PublicationTarget, + hooks: PublicationHooks, + db: Ledger = productionLedger +): Promise { + let state: ledger.PublicationStateSnapshot; + try { + await db.assertPublicationLedgerContract(); + state = await db.loadPublicationStateForQueueItem(target.userId, target.queueItemId); + } catch { + return result(target, 'blocked', 'publication_ledger_schema_unavailable'); + } + const existing = terminalResult(target, state); + if (existing) return existing; + // This executor is hosted-only. Do not claim/freeze unavailable provider media, + // refresh credentials, or spend on generation merely to rediscover retirement. + if (target.platform === 'threads' || target.platform === 'instagram') { + return result(target, 'blocked', 'legacy_meta_publication_disabled', state); + } + if (target.platform === 'facebook') return result(target, 'blocked', 'facebook_paused', state); + + const claimToken = ledger.createPublicationClaimToken(); + let intent: ledger.PublicationIntent; + try { + intent = await db.claimPublicationIntent(target.userId, target.queueItemId, claimToken, 120); + } catch { + const current = await readBestEffort(target, db); + return terminalResult(target, current) || result(target, 'blocked', 'publication_claim_not_acquired', current); + } + // Do not release a row with a mismatched identity, even when the Data API returned it. + if (!validIntent(target, intent) || intent.claim_token !== claimToken + || intent.state !== 'claimed' || !Number.isSafeInteger(intent.claim_version)) { + return result(target, 'unknown', 'publication_identity_mismatch'); + } + const payload = Object.freeze({ ...intent.payload }); + let prepared: PreparedPublication; + try { + if (typeof payload.text !== 'string' || !payload.text.trim()) throw new PublicationPreflightError('draft_text_missing'); + if (!Number.isFinite(Date.parse(intent.claim_expires_at || '')) + || Date.parse(intent.claim_expires_at || '') <= Date.now()) throw new PublicationPreflightError('publication_claim_expired'); + prepared = await hooks.prepare(payload); + if (!prepared.providerAccountRef?.trim()) throw new PublicationPreflightError('provider_account_identity_required'); + } catch (error) { + const code = error instanceof PublicationPreflightError ? error.code : 'publication_preflight_failed'; + try { + await db.releasePublicationClaim(target.userId, intent, code); + } catch { + // The lease may have expired or another owner may have taken it. Never force status. + } + return result(target, 'blocked', code, { intent }); + } + + const operationId = ledger.createDispatchOperationId(); + let attempt: ledger.PublicationAttempt; + try { + attempt = await db.beginPublicationDispatch({ + userId: target.userId, + intent, + dispatchOperationId: operationId, + providerAccountRef: prepared.providerAccountRef, + providerIdempotencySupported: false, + }); + } catch { + // The begin RPC may have committed. Never release the claim or send blindly. + const current = await readBestEffort(target, db); + if (validAttempt(target, current) && current.attempt?.dispatch_operation_id === operationId) { + return recordUnknownBestEffort(target, intent, current.attempt, 'publication_dispatch_receipt_lost', db); + } + return result(target, 'unknown', 'publication_dispatch_receipt_lost', { intent }); + } + if (!validAttempt(target, { intent, attempt }) || attempt.dispatch_operation_id !== operationId) { + return result(target, 'unknown', 'publication_identity_mismatch'); + } + if (attempt.state !== 'dispatching') { + return terminalResult(target, { intent, attempt }) || result(target, 'unknown', 'publication_dispatch_not_authorised', { intent, attempt }); + } + + let acceptance: ProviderAcceptance; + try { + acceptance = await prepared.send(); + if (typeof acceptance.externalPostId !== 'string' || !acceptance.externalPostId.trim() + || acceptance.externalPostId === 'posted') throw new Error('provider receipt missing'); + } catch (error) { + const classification = classifyPostDispatchError(error); + if (classification.outcome === 'rejected') { + try { + const rejected = await db.recordPublicationRejected({ + userId: target.userId, + attempt, + errorCode: classification.code, + errorMessage: classification.message, + providerReceipt: classification.providerReceipt, + }); + return result(target, 'rejected', classification.code, { intent, attempt: rejected }); + } catch { + // A known external rejection with failed local persistence is still non-retryable. + } + } + return recordUnknownBestEffort(target, intent, attempt, classification.code, db); + } + + // This is deliberately outside the provider catch. Database errors are not provider rejections. + let accepted: ledger.PublicationAttempt; + try { + accepted = await db.recordPublicationAccepted({ + userId: target.userId, + attempt, + externalPostId: acceptance.externalPostId, + providerReceipt: { external_post_id: acceptance.externalPostId }, + }); + if (!validAttempt(target, { intent, attempt: accepted }) + || !['accepted', 'verified'].includes(accepted.state) + || accepted.external_post_id !== acceptance.externalPostId) throw new Error('acceptance receipt mismatch'); + } catch { + return recordUnknownBestEffort(target, intent, attempt, 'publication_acceptance_record_pending', db, acceptance.externalPostId); + } + let bookkeepingPending = false; + try { + await hooks.afterAccepted?.(accepted, payload); + } catch { + bookkeepingPending = true; + } + // Enrich the response with exact history identity without making this read a + // condition of provider acceptance or reopening the send path on read failure. + const finalState = await readBestEffort(target, db); + const receiptState = validAttempt(target, finalState) + && finalState.attempt?.id === accepted.id + && finalState.attempt.external_post_id === accepted.external_post_id + ? finalState : { intent, attempt: accepted }; + const finished = result(target, accepted.state === 'verified' ? 'verified' : 'accepted', null, receiptState, { bookkeepingPending }); + if (bookkeepingPending) finished.jobStatus = 'completed_with_errors'; + return finished; +} + +/** Recovery never dispatches. Database identity, not logs or source URLs, determines the outcome. */ +export async function reconcilePublication( + target: PublicationTarget, + now = Date.now(), + db: Ledger = productionLedger +): Promise { + let state: ledger.PublicationStateSnapshot; + try { + await db.assertPublicationLedgerContract(); + state = await db.loadPublicationStateForQueueItem(target.userId, target.queueItemId); + } catch { + return result(target, 'unknown', 'publication_reconciliation_unavailable'); + } + if (!state.intent) return result(target, 'unknown', 'legacy_publication_requires_reconciliation'); + if (!validAttempt(target, state)) return result(target, 'unknown', 'publication_identity_mismatch'); + const { intent, attempt } = state; + if (attempt?.state === 'dispatching' && Date.parse(attempt.dispatch_started_at) <= now - 180_000) { + return recordUnknownBestEffort(target, intent, attempt, 'publication_dispatch_interrupted', db); + } + const terminal = terminalResult(target, state); + if (terminal) return terminal; + if (intent.state === 'claimed') { + if (!(Date.parse(intent.claim_expires_at || '') <= now)) return result(target, 'blocked', 'publication_claim_active', state); + try { + await db.releasePublicationClaim(target.userId, intent, 'publication_predispatch_lease_expired'); + } catch { + const current = await readBestEffort(target, db); + return terminalResult(target, current) || result(target, 'blocked', 'publication_recovery_raced'); + } + } + return result(target, 'retry_wait', 'publication_not_dispatched', state); +} + +/** Also recovers orphaned publish_all attempts after their parent job has already ended. */ +export async function recoverStalePublications(now = Date.now()): Promise { + try { + await ledger.assertPublicationLedgerContract(); + const intents = await supabaseSelect('publication_intents', { + filters: [ + { column: 'state', operator: 'in', value: ['claimed', 'dispatching'] }, + { column: 'updated_at', operator: 'lte', value: new Date(now - 180_000).toISOString() }, + ], + order: 'updated_at.asc', + limit: 50, + }); + let checked = 0; + for (const intent of intents) { + try { + await reconcilePublication({ userId: intent.user_id, queueItemId: intent.queue_item_id, platform: intent.platform }, now); + checked++; + } catch { + // One malformed/blocked tenant cannot stop other recovery or ready-post draining. + } + } + return checked; + } catch { + return 0; + } +} diff --git a/src/publication-ledger.ts b/src/publication-ledger.ts new file mode 100644 index 0000000..4fa5d92 --- /dev/null +++ b/src/publication-ledger.ts @@ -0,0 +1,335 @@ +import { randomUUID } from 'node:crypto'; + +import { + SupabaseRestError, + supabaseRpc, + supabaseSelect, +} from './supabase-client'; +import type { PlatformKey } from './types'; + +export const PUBLICATION_SCHEMA_CONTRACT = 'publication-ledger-v1'; +export const PUBLICATION_SCHEMA_MIGRATION = '20260907054000'; +export const REQUIRED_PUBLICATION_CAPABILITIES = [ + 'publication-intent-claim-v1', + 'publication-dispatch-boundary-v1', + 'publication-attempt-outcome-v1', + 'publication-unknown-reconciliation-v1', + 'publication-exact-history-receipt-v1', + 'publication-queue-compatibility-fence-v1', + 'publication-provenance-snapshot-v1', +] as const; + +export interface PublicationSchemaContract { + contract: string; + migration: string; + capabilities: string[]; +} + +export interface PublicationPayload { + platform: PlatformKey; + text: string; + instagram_image_url?: string; + source_url?: string; + source_title?: string; + angle?: string; + angle_record_id?: string; +} + +export type PublicationIntentState = + | 'scheduled' + | 'claimed' + | 'dispatching' + | 'accepted' + | 'rejected' + | 'unknown' + | 'verified'; + +export type PublicationAttemptState = + | 'dispatching' + | 'accepted' + | 'rejected' + | 'unknown' + | 'verified'; + +export interface PublicationIntent { + id: string; + user_id: string; + queue_item_id: string; + platform: PlatformKey; + queue_status_before_claim: string; + queue_revision_updated_at: string; + payload: PublicationPayload; + state: PublicationIntentState; + claim_token?: string | null; + claim_version: number; + claim_expires_at?: string | null; + created_at: string; + updated_at: string; +} + +export interface PublicationAttempt { + id: string; + intent_id: string; + user_id: string; + queue_item_id: string; + platform: PlatformKey; + attempt_no: number; + dispatch_operation_id: string; + state: PublicationAttemptState; + provider_account_ref?: string | null; + provider_idempotency_key?: string | null; + provider_idempotency_supported: boolean; + external_post_id?: string | null; + external_url?: string | null; + provider_published_at?: string | null; + provider_receipt?: Record | null; + error_code?: string | null; + error_message?: string | null; + reconciliation_evidence?: Record | null; + verification_evidence?: Record | null; + dispatch_started_at: string; + outcome_recorded_at?: string | null; + reconciled_at?: string | null; + verified_at?: string | null; + created_at: string; + updated_at: string; +} + +export interface PublicationHistoryReceipt { + id: string; + user_id: string; + platform: PlatformKey; + post_text?: string | null; + external_post_id?: string | null; + external_url?: string | null; + source_url?: string | null; + published_at: string; + queue_item_id?: string | null; + publication_intent_id?: string | null; + publication_attempt_id?: string | null; +} + +export interface PublicationStateSnapshot { + intent?: PublicationIntent; + attempt?: PublicationAttempt; + history?: PublicationHistoryReceipt; +} + +export class PublicationLedgerContractError extends Error { + readonly code = 'publication_ledger_schema_unavailable'; + + constructor(message: string, public readonly causeDetails?: string) { + super(message); + } +} + +export function createPublicationClaimToken(): string { + return randomUUID(); +} + +export function createDispatchOperationId(): string { + return randomUUID(); +} + +export async function getPublicationSchemaContract(): Promise { + return supabaseRpc('get_publication_schema_contract', {}, { retrySafe: true }); +} + +export async function assertPublicationLedgerContract(): Promise { + try { + const contract = await getPublicationSchemaContract(); + if (contract?.contract !== PUBLICATION_SCHEMA_CONTRACT) { + throw new PublicationLedgerContractError( + `Expected ${PUBLICATION_SCHEMA_CONTRACT}, received ${String(contract?.contract || 'missing')}` + ); + } + if (contract.migration !== PUBLICATION_SCHEMA_MIGRATION) { + throw new PublicationLedgerContractError( + `Expected ${PUBLICATION_SCHEMA_CONTRACT} migration ${PUBLICATION_SCHEMA_MIGRATION}, received ${String(contract.migration || 'missing')}` + ); + } + const capabilities = new Set(contract.capabilities || []); + const missing = REQUIRED_PUBLICATION_CAPABILITIES.filter(capability => !capabilities.has(capability)); + if (missing.length) { + throw new PublicationLedgerContractError( + `${PUBLICATION_SCHEMA_CONTRACT} is missing required capabilities: ${missing.join(', ')}` + ); + } + return contract; + } catch (error) { + if (error instanceof PublicationLedgerContractError) throw error; + const detail = error instanceof SupabaseRestError + ? `HTTP ${error.status}` + : error instanceof Error + ? error.name + : 'unknown'; + throw new PublicationLedgerContractError( + `${PUBLICATION_SCHEMA_CONTRACT} is not available on the configured Supabase project`, + detail + ); + } +} + +export function isPublicationLedgerSchemaUnavailable(error: unknown): boolean { + return error instanceof PublicationLedgerContractError; +} + +export async function claimPublicationIntent( + userId: string, + queueItemId: string, + claimToken = createPublicationClaimToken(), + leaseSeconds = 120 +): Promise { + const rows = await supabaseRpc('claim_publication_intent', { + p_user_id: userId, + p_queue_item_id: queueItemId, + p_claim_token: claimToken, + p_lease_seconds: leaseSeconds, + }, { retrySafe: true }); + const intent = rows[0]; + if (!intent) throw new Error('claim_publication_intent returned no intent'); + return intent; +} + +export async function releasePublicationClaim( + userId: string, + intent: Pick, + reason: string +): Promise { + if (!intent.claim_token) throw new Error('publication claim token missing'); + const rows = await supabaseRpc('release_publication_claim', { + p_user_id: userId, + p_intent_id: intent.id, + p_claim_token: intent.claim_token, + p_claim_version: intent.claim_version, + p_reason: reason, + }, { retrySafe: true }); + const released = rows[0]; + if (!released) throw new Error('release_publication_claim returned no intent'); + return released; +} + +export interface BeginDispatchInput { + userId: string; + intent: Pick; + dispatchOperationId?: string; + providerAccountRef?: string | null; + providerIdempotencyKey?: string | null; + providerIdempotencySupported?: boolean; +} + +export async function beginPublicationDispatch(input: BeginDispatchInput): Promise { + if (!input.intent.claim_token) throw new Error('publication claim token missing'); + const dispatchOperationId = input.dispatchOperationId || createDispatchOperationId(); + const rows = await supabaseRpc('begin_publication_dispatch', { + p_user_id: input.userId, + p_intent_id: input.intent.id, + p_claim_token: input.intent.claim_token, + p_claim_version: input.intent.claim_version, + p_dispatch_operation_id: dispatchOperationId, + p_provider_account_ref: input.providerAccountRef ?? null, + p_provider_idempotency_key: input.providerIdempotencyKey ?? null, + p_provider_idempotency_supported: input.providerIdempotencySupported === true, + }, { retrySafe: true }); + const attempt = rows[0]; + if (!attempt) throw new Error('begin_publication_dispatch returned no attempt'); + return attempt; +} + +export interface PublicationAcceptedInput { + userId: string; + attempt: Pick; + externalPostId: string; + externalUrl?: string | null; + providerPublishedAt?: string | null; + providerReceipt?: Record | null; +} + +export async function recordPublicationAccepted(input: PublicationAcceptedInput): Promise { + const rows = await supabaseRpc('record_publication_accepted', { + p_user_id: input.userId, + p_attempt_id: input.attempt.id, + p_dispatch_operation_id: input.attempt.dispatch_operation_id, + p_external_post_id: input.externalPostId, + p_external_url: input.externalUrl ?? null, + p_provider_published_at: input.providerPublishedAt ?? null, + p_provider_receipt: input.providerReceipt ?? null, + }, { retrySafe: true }); + const attempt = rows[0]; + if (!attempt) throw new Error('record_publication_accepted returned no attempt'); + return attempt; +} + +export interface PublicationFailureInput { + userId: string; + attempt: Pick; + errorCode?: string | null; + errorMessage?: string | null; + providerReceipt?: Record | null; +} + +export async function recordPublicationRejected(input: PublicationFailureInput): Promise { + const rows = await supabaseRpc('record_publication_rejected', { + p_user_id: input.userId, + p_attempt_id: input.attempt.id, + p_dispatch_operation_id: input.attempt.dispatch_operation_id, + p_error_code: input.errorCode ?? null, + p_error_message: input.errorMessage ?? null, + p_provider_receipt: input.providerReceipt ?? null, + }, { retrySafe: true }); + const attempt = rows[0]; + if (!attempt) throw new Error('record_publication_rejected returned no attempt'); + return attempt; +} + +export async function recordPublicationUnknown(input: PublicationFailureInput): Promise { + const rows = await supabaseRpc('record_publication_unknown', { + p_user_id: input.userId, + p_attempt_id: input.attempt.id, + p_dispatch_operation_id: input.attempt.dispatch_operation_id, + p_error_code: input.errorCode ?? null, + p_error_message: input.errorMessage ?? null, + p_provider_receipt: input.providerReceipt ?? null, + }, { retrySafe: true }); + const attempt = rows[0]; + if (!attempt) throw new Error('record_publication_unknown returned no attempt'); + return attempt; +} + +export async function loadPublicationStateForQueueItem( + userId: string, + queueItemId: string +): Promise { + const intent = (await supabaseSelect('publication_intents', { + select: '*', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'queue_item_id', operator: 'eq', value: queueItemId }, + ], + limit: 1, + }))[0]; + if (!intent) return {}; + + const attempt = (await supabaseSelect('publication_attempts', { + select: '*', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'intent_id', operator: 'eq', value: intent.id }, + ], + order: 'attempt_no.desc', + limit: 1, + }))[0]; + + const history = attempt + ? (await supabaseSelect('publish_history', { + select: 'id,user_id,platform,post_text,external_post_id,external_url,source_url,published_at,queue_item_id,publication_intent_id,publication_attempt_id', + filters: [ + { column: 'user_id', operator: 'eq', value: userId }, + { column: 'publication_attempt_id', operator: 'eq', value: attempt.id }, + ], + limit: 1, + }))[0] + : undefined; + + return { intent, attempt, history }; +} diff --git a/src/publication-outcome.ts b/src/publication-outcome.ts new file mode 100644 index 0000000..d12d1ad --- /dev/null +++ b/src/publication-outcome.ts @@ -0,0 +1,34 @@ +import { isPlatformPublishError } from './platform-errors'; + +export type PostDispatchOutcome = 'rejected' | 'unknown'; + +export interface PublicationErrorClassification { + outcome: PostDispatchOutcome; + code: string; + message: string; + providerReceipt: Record; +} + +// A conflict may describe an existing post; a timeout is never proof of absence. +const DEFINITIVE_REJECTION_STATUSES = new Set([400, 401, 403, 404, 413, 415, 422, 429]); + +export function classifyPostDispatchError(error: unknown): PublicationErrorClassification { + const providerError = isPlatformPublishError(error) ? error : undefined; + const rejected = Boolean(providerError?.stage === 'post' + && DEFINITIVE_REJECTION_STATUSES.has(providerError.status || 0)); + const safeCode = providerError && /^[a-z0-9_]{1,80}$/.test(providerError.code) + ? providerError.code : 'unclassified_error'; + const outcome = rejected ? 'rejected' : 'unknown'; + return { + outcome, + code: `provider_${rejected ? 'rejected' : 'outcome_unknown'}_${safeCode}`, + message: rejected + ? 'The provider rejected this publication attempt.' + : 'The provider outcome is uncertain. Reconcile the exact attempt before retrying.', + // Deliberate allowlist: never persist raw errors, headers, URLs, or response snippets. + providerReceipt: { + ...(providerError ? { platform: providerError.platform, http_status: providerError.status ?? null } : {}), + outcome_classification: rejected ? 'known_rejected' : 'ambiguous_unknown', + }, + }; +} diff --git a/src/publication-receipts.ts b/src/publication-receipts.ts new file mode 100644 index 0000000..e5495e3 --- /dev/null +++ b/src/publication-receipts.ts @@ -0,0 +1,36 @@ +import { recordPublicationAccepted, type PublicationAcceptedInput, type PublicationAttempt } from './publication-ledger'; +import { SupabaseRestError, supabaseRpc } from './supabase-client'; + +/** + * A delayed successful provider response is positive evidence for this exact + * attempt. It may race the stale-attempt scanner, but it never authorises a resend. + * A negative provider search is deliberately not an input to this operation. + */ +export async function recordObservedAcceptance(input: PublicationAcceptedInput): Promise { + const observedAt = new Date().toISOString(); + try { + return await recordPublicationAccepted(input); + } catch (error) { + if (!(error instanceof SupabaseRestError) + || error.message !== 'publication_unknown_requires_reconciliation') throw error; + } + const rows = await supabaseRpc('resolve_publication_unknown_accepted', { + p_user_id: input.userId, + p_attempt_id: input.attempt.id, + p_dispatch_operation_id: input.attempt.dispatch_operation_id, + p_external_post_id: input.externalPostId, + p_external_url: input.externalUrl ?? null, + p_provider_published_at: input.providerPublishedAt ?? null, + p_provider_receipt: input.providerReceipt ?? null, + p_reconciliation_evidence: { + kind: 'exact_attempt_provider_acceptance_response', + attempt_id: input.attempt.id, + dispatch_operation_id: input.attempt.dispatch_operation_id, + external_post_id: input.externalPostId, + observed_at: observedAt, + }, + }, { retrySafe: true }); + const accepted = rows[0]; + if (!accepted) throw new Error('positive provider receipt reconciliation returned no attempt'); + return accepted; +} diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index bce06f9..c436436 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -9,6 +9,7 @@ import * as linkedin from './linkedin'; import * as logger from './logger'; import { activePlatformsFromSettings } from './platform-settings'; import * as workerClaims from './worker-claims'; +import { executePublication, reconcilePublication, recoverStalePublications, PublicationPreflightError } from './publication-executor'; import * as threads from './threads'; import * as x from './x'; import { buildDailyInventoryPlan, type DailyInventoryQueueRow } from './daily-inventory-planner'; @@ -403,7 +404,7 @@ const PUBLISH_UNKNOWN_STATE_CODE = 'unknown_publish_state'; const PUBLISH_UNKNOWN_STATE_MESSAGE = 'Scheduled publish timed out before the system could confirm whether the platform accepted the post. Review the platform account before retrying.'; const PUBLISH_UNKNOWN_STATE_NEXT_ACTION = - 'Check the platform account for a matching post. If it is not live, retry this queue item manually.'; + 'Do not retry or recreate this post. Reconcile the exact publication attempt first; absence from a search is not proof of rejection.'; const PUBLISH_INTERRUPTED_CODE = 'publish_claim_interrupted'; const PUBLISH_INTERRUPTED_MESSAGE = 'Scheduled publish was interrupted before the platform publish step started. The queue item remains available for retry.'; @@ -3588,139 +3589,81 @@ async function assertOpenAIRepairAllowedForPublish( }); } -async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, settings: UserSettingsRow): Promise { - const locked = await supabaseUpdate('queue_items', { - status: 'publishing', - error_message: null, - }, { - filters: [ - { column: 'id', operator: 'eq', value: row.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ['pending', 'ready', 'failed'] }, - ], - returning: true, - }); - - let current = locked[0]; - if (!current) { - throw new WorkerJobError('queue_item_not_available', 'queue_item_not_available', { queueItemId: row.id }); - } - - try { - if (publishRequiresOpenAIMediaRepair(current)) { - await assertOpenAIRepairAllowedForPublish(job, current, settings); - const image = await ai.ensurePersistentInstagramImage({ - imageUrl: current.instagram_image_url, - imagePrompt: current.instagram_image_prompt, - title: current.source_title || current.angle || 'Instagram post', - text: current.draft_text || '', - }, { - angleId: current.angle_record_id || undefined, - jobId: job.id, - jobKind: job.kind, - platform: current.platform, - queueItemId: current.id, - stage: 'instagram_publish_media_repair', - userId: job.user_id, - }); - const patched = await supabaseUpdate('queue_items', { - instagram_image_url: image.imageUrl, - instagram_image_prompt: current.instagram_image_prompt || image.imagePrompt, - }, { - filters: [ - { column: 'id', operator: 'eq', value: current.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - returning: true, - }); - current = patched[0] || { - ...current, - instagram_image_url: image.imageUrl, - instagram_image_prompt: current.instagram_image_prompt || image.imagePrompt, - }; - } - - if (current.platform === 'threads') { - await prepareThreadsCredentialForPublish(job.user_id); - } - if (current.platform === 'linkedin') { - await refreshLinkedInCredentialForPublish(job.user_id); - } - const authMode = current.platform === 'x' - ? await verifyXCredentialForPublish(job.user_id) - : undefined; - const externalPostId = await publishPlatform(current); - if (current.platform === 'linkedin') { - await markLinkedInCredentialVerified(job.user_id); - } - await supabaseUpdate('queue_items', { - status: 'published', - error_message: null, - }, { - filters: [ - { column: 'id', operator: 'eq', value: current.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - const publishedAt = nowIso(); - const publishHistoryRows = await supabaseInsert('publish_history', { - user_id: job.user_id, - platform: current.platform, - post_text: current.draft_text || null, - external_post_id: externalPostId, - source_url: current.source_url || null, - published_at: publishedAt, - }, true); - const publishHistory = publishHistoryRows[0]; - if (current.angle_record_id) { - await supabaseUpdate('angle_records', { - status: 'published', - last_used_at: nowIso(), - }, { - filters: [ - { column: 'id', operator: 'eq', value: current.angle_record_id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], - }); - } - await writeWorkerLog(job.user_id, 'info', 'published_queue_item', { - jobId: job.id, - queueItemId: current.id, - platform: current.platform, - ...(authMode ? { auth_mode: authMode } : {}), - externalPostId, - publishHistoryId: publishHistory?.id || null, - }); - return publishSuccessResult(job, current, publishHistory, externalPostId, nowIso(), authMode); - } catch (error) { - const message = publicError(error); - const imageErrorContext = safeErrorContext(error); - if (current.platform === 'instagram' && imageErrorContext?.stage === ai.OPENAI_IMAGE_GENERATION_STAGE) { - await writeWorkerLog(job.user_id, 'warn', 'instagram_image_generation_failed', { - jobId: job.id, - queueItemId: current.id, - platform: current.platform, - ...imageErrorContext, - }); - } - if (isPlatformPublishError(error)) { - await writeWorkerLog(job.user_id, 'warn', 'platform_publish_failed', { - jobId: job.id, - queueItemId: current.id, - ...platformErrorContext(error), - }); - } - await supabaseUpdate('queue_items', { - status: 'failed', - error_message: message, +async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: UserSettingsRow): Promise { + if (row.user_id !== job.user_id) throw new WorkerJobError('publication_tenant_mismatch'); + // Reload per item. publish_all must not retain the first item's stale credentials/settings. + const tenant = await loadTenantContext(job.user_id); + return withTenantRuntime(tenant, async () => { + const execution = await executePublication({ + userId: job.user_id, queueItemId: row.id, platform: row.platform, }, { - filters: [ - { column: 'id', operator: 'eq', value: current.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - ], + prepare: async payload => { + // Disabled hosted publishers are rejected before token refresh or paid media work. + if (payload.platform === 'threads' || payload.platform === 'instagram') { + throw new PublicationPreflightError('legacy_meta_publication_disabled'); + } + if (payload.platform === 'facebook') throw new PublicationPreflightError('facebook_paused'); + let providerAccountRef: string; + if (payload.platform === 'x') { + if (payload.text.trim().length > 280) throw new PublicationPreflightError('x_text_too_long'); + const verification = await x.verifyCredentials(); + providerAccountRef = verification.accountId; + } else { + if (!config.LINKEDIN_TOKEN || !config.LINKEDIN_PERSON_URN) { + throw new PublicationPreflightError('linkedin_not_connected'); + } + await refreshLinkedInCredentialForPublish(job.user_id); + providerAccountRef = config.LINKEDIN_PERSON_URN; + } + // Recheck entitlement/pause/enablement immediately before the dispatch boundary. + const entitlement = await loadEntitlement(job.user_id); + if (!entitlement.canWrite) throw new PublicationPreflightError('billing_inactive'); + const settings = (await supabaseSelect('user_settings', { + filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }], limit: 1, + }))[0] || {}; + if (!activePlatformsFromSettings(settings).includes(payload.platform)) { + throw new PublicationPreflightError('platform_disabled'); + } + if (jobOrigin(job) === 'scheduled') { + if (settings.automation_enabled !== true || settings.automation_publish_enabled !== true) { + throw new PublicationPreflightError('publish_automation_disabled'); + } + // Schedule is frozen by the claimed intent. Content still comes only from payload. + const frozenQueue = (await supabaseSelect('queue_items', { + select: 'id,user_id,scheduled_for', + filters: [{ column: 'id', operator: 'eq', value: row.id }, { column: 'user_id', operator: 'eq', value: job.user_id }], + limit: 1, + }))[0]; + if (!frozenQueue || !(Date.parse(frozenQueue.scheduled_for) <= Date.now())) { + throw new PublicationPreflightError('publication_not_due'); + } + } + const frozenRow: QueueItemRow = { + ...row, platform: payload.platform, draft_text: payload.text, + instagram_image_url: payload.instagram_image_url || null, + source_url: payload.source_url || null, source_title: payload.source_title || null, + angle: payload.angle || null, angle_record_id: payload.angle_record_id || null, + }; + return { + providerAccountRef, + send: async () => ({ externalPostId: await publishPlatform(frozenRow) }), + }; + }, + afterAccepted: async (_attempt, payload) => { + if (payload.angle_record_id) { + await supabaseUpdate('angle_records', { status: 'published', last_used_at: nowIso() }, { + filters: [{ column: 'id', operator: 'eq', value: payload.angle_record_id }, { column: 'user_id', operator: 'eq', value: job.user_id }], + }); + } + }, }); - throw error; - } + // Non-authoritative telemetry cannot change publication truth or trigger a resend. + try { + await writeWorkerLog(job.user_id, execution.outcome === 'accepted' || execution.outcome === 'verified' ? 'info' : 'warn', + 'publication_result', { jobId: job.id, ...execution }); + } catch { /* Durable ledger remains the source of truth. */ } + return execution; + }); } async function handlePublishNow(job: AgentJobRow, tenant: TenantContext): Promise { @@ -3741,37 +3684,25 @@ async function handlePublishNow(job: AgentJobRow, tenant: TenantContext): Promis async function handlePublishAll(job: AgentJobRow, tenant: TenantContext): Promise { const rows = await supabaseSelect('queue_items', { - select: '*', - filters: [ - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'in', value: ['pending', 'ready'] }, - ], - order: 'scheduled_for.asc', - limit: 100, + filters: [{ column: 'user_id', operator: 'eq', value: job.user_id }, { column: 'status', operator: 'in', value: ['pending', 'ready'] }], + order: 'scheduled_for.asc', limit: 100, }); - const published: JsonMap[] = []; const failures: JsonMap[] = []; for (const row of rows) { try { - published.push(await publishQueueRow(job, row, tenant.settings)); - } catch (error) { - failures.push({ - queueItemId: row.id, - platform: row.platform, - error: publicError(error), - }); + const outcome = await publishQueueRow(job, row, tenant.settings); + if (outcome.outcome === 'accepted' || outcome.outcome === 'verified') published.push(outcome); + else failures.push(outcome); + } catch { + failures.push({ queueItemId: row.id, platform: row.platform, failureCode: 'publication_execution_interrupted', nextAction: 'Reconcile the exact publication before retrying.' }); } } - - if (failures.length) { - throw new WorkerJobError('publish_all_failed', 'publish_all_failed', { - published, - failures, - }); - } - - return { published, failures }; + const outcome = failures.length ? (published.length ? 'completed_with_errors' : 'blocked') : 'accepted'; + return { published, failures, outcome, + jobStatus: failures.length ? (published.length ? 'completed_with_errors' : 'failed') : 'completed', + summary: { outcome, published, failures, errors: failures.map(item => item.failureCode) }, + }; } async function handleSkipSlot(job: AgentJobRow): Promise { @@ -4750,17 +4681,6 @@ function hasRecentJobActivity(logs: WorkerLogRow[], cutoffIso: string): boolean }); } -function publishStageStarted(logs: WorkerLogRow[], queueItemId: string): boolean { - return logs.some(row => { - const context = logContext(row); - const loggedQueueItemId = logString(context, 'queueItemId'); - if (loggedQueueItemId && loggedQueueItemId !== queueItemId) return false; - return row.message === 'published_queue_item' - || row.message === 'platform_publish_failed' - || row.message === 'instagram_image_generation_failed'; - }); -} - async function loadQueueItemForStalePublish(job: AgentJobRow, queueItemId: string): Promise { return (await supabaseSelect('queue_items', { select: 'id,user_id,platform,status,slot_index,scheduled_for,source_url,source_title,angle,error_message', @@ -4772,144 +4692,16 @@ async function loadQueueItemForStalePublish(job: AgentJobRow, queueItemId: strin }))[0]; } -async function findPublishHistoryForQueueItem( - job: AgentJobRow, - row: QueueItemRow -): Promise { - if (!row.source_url) return undefined; - return (await supabaseSelect('publish_history', { - select: 'id,user_id,platform,external_post_id,external_url,source_url,published_at', - filters: [ - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'platform', operator: 'eq', value: row.platform }, - { column: 'source_url', operator: 'eq', value: row.source_url }, - { column: 'published_at', operator: 'gte', value: job.started_at || job.created_at }, - ], - order: 'published_at.desc', - limit: 1, - }))[0]; -} +// Legacy source-URL/time history matching removed. Ledger recovery uses exact identities. -function stalePublishResult( - job: AgentJobRow, - row: QueueItemRow | undefined, - history: PublishHistoryRow | undefined, - logs: WorkerLogRow[] -): JsonMap { +async function stalePublishJobResult(job: AgentJobRow, _logs: WorkerLogRow[]): Promise { const queueItemId = queueItemIdFromPayload(job.payload); - if (!row) { - const message = 'Scheduled publish timed out, but the queue item no longer exists.'; - const nextAction = 'Review Logs and publish history before retrying scheduled publishing.'; - return { - outcome: 'blocked', - message, - nextAction, - error: 'queue_item_missing', - jobStatus: 'failed', - summary: { - outcome: 'blocked', - message, - nextAction, - failedStage: 'scheduled_publish', - failureCode: 'queue_item_missing', - queueItemId, - queueItemStatus: 'missing', - errors: ['queue_item_missing'], - }, - }; - } - const platform = row?.platform || 'unknown'; - const externalPostId = String(history?.external_post_id || '').trim(); - if (row.status === 'published' && history?.id && externalPostId) { - return publishSuccessResult( - job, - row, - history, - externalPostId, - history.published_at || nowIso() - ); - } - const reconciled = row?.status === 'published' || Boolean(history); - const stageStarted = row?.status === 'publishing' || (queueItemId ? publishStageStarted(logs, queueItemId) : false); - const code = reconciled - ? PUBLISH_RECONCILED_CODE - : stageStarted - ? PUBLISH_UNKNOWN_STATE_CODE - : PUBLISH_INTERRUPTED_CODE; - const message = reconciled - ? PUBLISH_RECONCILED_MESSAGE - : stageStarted - ? PUBLISH_UNKNOWN_STATE_MESSAGE - : PUBLISH_INTERRUPTED_MESSAGE; - const nextAction = reconciled - ? PUBLISH_RECONCILED_NEXT_ACTION - : stageStarted - ? PUBLISH_UNKNOWN_STATE_NEXT_ACTION - : PUBLISH_INTERRUPTED_NEXT_ACTION; - const outcome = reconciled ? 'completed_with_errors' : 'blocked'; - - const summary: JsonMap = { - outcome, - message, - nextAction, - failedStage: reconciled ? 'publish_state_reconciliation' : stageStarted ? 'scheduled_publish' : 'publish_claim', - failureCode: code, - platform, - queueItemId, - queueItemStatus: row?.status || 'missing', - scheduledFor: row?.scheduled_for || null, - publishHistoryId: history?.id || null, - externalPostId: history?.external_post_id || null, - errors: [message], - }; - - return { - outcome, - message, - nextAction, - error: code, - jobStatus: reconciled ? 'completed_with_errors' : 'failed', - summary, + const row = queueItemId ? await loadQueueItemForStalePublish(job, queueItemId) : undefined; + if (!row) return { + outcome: 'unknown', jobStatus: 'failed', failureCode: 'publication_queue_identity_missing', + summary: { outcome: 'unknown', failureCode: 'publication_queue_identity_missing', nextAction: 'Reconcile publication identity before retrying. Do not recreate the post.', errors: ['publication_queue_identity_missing'] }, }; -} - -async function stalePublishJobResult(job: AgentJobRow, logs: WorkerLogRow[]): Promise { - const queueItemId = queueItemIdFromPayload(job.payload); - if (!queueItemId) { - return { - outcome: 'blocked', - message: 'Scheduled publish timed out and did not include a queue item id.', - nextAction: 'Review the job payload before retrying scheduled publishing.', - error: 'missing_queue_item_id', - jobStatus: 'failed', - summary: { - outcome: 'blocked', - message: 'Scheduled publish timed out and did not include a queue item id.', - nextAction: 'Review the job payload before retrying scheduled publishing.', - failedStage: 'scheduled_publish', - failureCode: 'missing_queue_item_id', - errors: ['missing_queue_item_id'], - }, - }; - } - - const row = await loadQueueItemForStalePublish(job, queueItemId); - const history = row ? await findPublishHistoryForQueueItem(job, row) : undefined; - const result = stalePublishResult(job, row, history, logs); - const summary = resultSummary(result); - if (row && summary.failureCode === PUBLISH_UNKNOWN_STATE_CODE) { - await supabaseUpdate('queue_items', { - status: 'failed', - error_message: PUBLISH_UNKNOWN_STATE_MESSAGE, - }, { - filters: [ - { column: 'id', operator: 'eq', value: row.id }, - { column: 'user_id', operator: 'eq', value: job.user_id }, - { column: 'status', operator: 'eq', value: 'publishing' }, - ], - }); - } - return result; + return reconcilePublication({ userId: job.user_id, queueItemId: row.id, platform: row.platform }); } async function cleanupStaleRunningJobs(stats: SchedulerStats, now: Date): Promise { @@ -4988,6 +4780,7 @@ export async function runSupabaseAutomationScheduler(): Promise skipped: {}, }; const now = new Date(); + await recoverStalePublications(now.getTime()); await cleanupStaleRunningJobs(stats, now); await enqueueDueFetchJobs(stats, now); await enqueueDueSlotFillJobs(stats, now); @@ -5096,6 +4889,9 @@ export function startSupabaseWorkerLoop(log = logger): { stop: () => void } | un } export const __test__ = { + publishQueueRow, + stalePublishJobResult, + handlePublishAll, buildOpenAIUsageDailySummary, contentStrategyPromptOptions, draftCreationPreflightForAngle, @@ -5122,7 +4918,6 @@ export const __test__ = { sourceIntentFor, sourceIntentRejectReasons, sourceScopeFor, - stalePublishResult, }; if (typeof require !== 'undefined' && typeof module !== 'undefined' && require.main === module) { diff --git a/src/x.ts b/src/x.ts index b670c7e..0cd7040 100644 --- a/src/x.ts +++ b/src/x.ts @@ -287,7 +287,8 @@ async function apiRequestWithAuth( if ( authMode === 'oauth2-user' && hasOAuth2RefreshConfig() - && classifyXError(response.message) === 'auth' + && method === 'GET' + && response.status === 401 ) { token = await refreshAndPersistOAuth2AccessToken(); response = await sendApiRequest(method, path, authMode, payload, token); @@ -309,7 +310,12 @@ async function apiRequestWithAuth( } if (response.status >= 400 || response.data.errors || response.data.error || response.data.detail) { - throw new Error('X API: ' + response.message); + throw new PlatformPublishError({ + platform: 'x', stage: payload ? 'post' : 'credential_check', + code: 'platform_api_error', status: response.status, + userMessage: 'X returned a non-success response.', + nextAction: 'Review the exact attempt before any further publish.', + }); } return response.data; diff --git a/test/provider-single-dispatch.test.ts b/test/provider-single-dispatch.test.ts new file mode 100644 index 0000000..9246290 --- /dev/null +++ b/test/provider-single-dispatch.test.ts @@ -0,0 +1,66 @@ +import assert from 'node:assert/strict'; +import config from '../config'; +import * as x from '../src/x'; +import * as linkedin from '../src/linkedin'; +import { PlatformPublishError } from '../src/platform-errors'; +import { classifyPostDispatchError } from '../src/publication-outcome'; + +async function main() { + const originalFetch = globalThis.fetch; + const original = { + X_OAUTH2_ACCESS_TOKEN: config.X_OAUTH2_ACCESS_TOKEN, + X_OAUTH2_REFRESH_TOKEN: config.X_OAUTH2_REFRESH_TOKEN, + X_CLIENT_ID: config.X_CLIENT_ID, X_CLIENT_SECRET: config.X_CLIENT_SECRET, + LINKEDIN_TOKEN: config.LINKEDIN_TOKEN, LINKEDIN_PERSON_URN: config.LINKEDIN_PERSON_URN, + }; + Object.assign(config, { + X_OAUTH2_ACCESS_TOKEN: 'fixture-access', X_OAUTH2_REFRESH_TOKEN: 'fixture-refresh', + X_CLIENT_ID: 'fixture-client', X_CLIENT_SECRET: 'fixture-secret', + LINKEDIN_TOKEN: 'fixture-linkedin', LINKEDIN_PERSON_URN: 'urn:li:person:fixture', + }); + const restore = x.setOAuth2TokenPersistence(async () => {}); + try { + for (const status of [401, 503, 201]) { + let posts = 0, refreshes = 0; + globalThis.fetch = (async (input, init) => { + const url = String(input); + if (url.includes('oauth2/token')) { refreshes++; return Response.json({ access_token: 'new-fixture', refresh_token: 'new-refresh' }); } + assert.ok(url.endsWith('/2/tweets')); + assert.equal(init?.method, 'POST'); posts++; + return Response.json({ title: 'Unauthorized', ...(status === 201 ? { data: { id: '123456' } } : {}) }, { status }); + }) as typeof fetch; + await assert.rejects(() => x.publish('fixture content')); + assert.equal(posts, 1); assert.equal(refreshes, 0); + console.log(`ok - X status ${status} cannot trigger an invisible second publishing request`); + } + let lookups = 0, refreshes = 0; + globalThis.fetch = (async input => { + const url = String(input); + if (url.includes('oauth2/token')) { refreshes++; return Response.json({ access_token: 'new-fixture', refresh_token: 'new-refresh' }); } + lookups++; + return lookups === 1 ? Response.json({ title: 'Unauthorized' }, { status: 401 }) : Response.json({ data: { id: 'fixture-account' } }); + }) as typeof fetch; + assert.equal((await x.verifyCredentials()).accountId, 'fixture-account'); + assert.equal(lookups, 2); assert.equal(refreshes, 1); + console.log('ok - read-only X identity verification can still refresh after explicit 401'); + + globalThis.fetch = (async () => new Response('', { status: 201, headers: { 'x-restli-id': 'urn:li:share:123456' } })) as typeof fetch; + assert.equal(await linkedin.publish('fixture content'), 'urn:li:share:123456'); + globalThis.fetch = (async () => new Response('', { status: 201 })) as typeof fetch; + await assert.rejects(() => linkedin.publish('fixture content')); + console.log('ok - LinkedIn uses its exact response identity and never fabricates posted'); + + for (const status of [408, 409, 500]) { + const outcome = classifyPostDispatchError(new PlatformPublishError({ + platform: 'x', stage: 'post', status, code: 'platform_api_error', + userMessage: 'refresh_token=TOP_SECRET', bodySnippet: 'Authorization: TOP_SECRET', nextAction: 'TOP_SECRET', + })); + assert.equal(outcome.outcome, 'unknown'); + assert.ok(!JSON.stringify(outcome).includes('TOP_SECRET')); + } + console.log('ok - ambiguous status and secret-bearing error payloads cannot become retry evidence or receipts'); + } finally { + restore(); globalThis.fetch = originalFetch; Object.assign(config, original); + } +} +main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/test/publication-database.integration.ts b/test/publication-database.integration.ts new file mode 100644 index 0000000..7f6ce04 --- /dev/null +++ b/test/publication-database.integration.ts @@ -0,0 +1,237 @@ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import config from '../config'; +import { encryptCredential } from '../src/tenant-credentials'; +import { installScopedConfig, runWithRuntimeScope } from '../src/runtime-scope'; +import { __test__ as worker } from '../src/supabase-worker'; +import { reconcilePublication, recoverStalePublications } from '../src/publication-executor'; +import { loadPublicationStateForQueueItem, claimPublicationIntent, beginPublicationDispatch } from '../src/publication-ledger'; + +// Real local database and Worker; social transport is intercepted. No live-provider +// authorisation or visibility claim is made by this suite. +if (process.env.PUBLICATION_DATABASE_TEST !== 'local-only') throw new Error('Local database test opt-in required'); +const local = JSON.parse(execFileSync('supabase', ['status', '--output', 'json'], { + cwd: '.schema-contract', encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], +})); +const api = new URL(local.API_URL); +if (!['127.0.0.1', 'localhost'].includes(api.hostname) || api.protocol !== 'http:') throw new Error('Refusing non-local Supabase'); +const names = execFileSync('docker', ['ps', '--format', '{{.Names}}'], { encoding: 'utf8' }).trim().split('\n').filter(name => /^supabase_db_[a-zA-Z0-9_-]+$/.test(name)); +assert.equal(names.length, 1, 'exactly one ephemeral local database'); +function sql(statement: string): string { + return execFileSync('docker', ['exec', '-i', names[0], 'psql', '-U', 'postgres', '-d', 'postgres', '-v', 'ON_ERROR_STOP=1', '-At'], { + input: statement, encoding: 'utf8', stdio: ['pipe', 'pipe', 'pipe'], + }).trim(); +} +const literal = (value: string) => `'${value.replace(/'/g, "''")}'`; +Object.assign(config, { + SUPABASE_URL: api.origin, SUPABASE_SERVICE_ROLE_KEY: local.SERVICE_ROLE_KEY, + CREDENTIAL_ENCRYPTION_KEY: 'ephemeral-publication-test-only', HTTP_TIMEOUT_MS: 5000, +}); +installScopedConfig(config); + +interface Fixture { + user: string; + row: { id: string; user_id: string; platform: 'x' | 'linkedin' | 'threads' | 'instagram'; slot_index: number; scheduled_for: string; status: string; draft_text: string }; + token: string; +} +const tokens = new Map(); +const posts = new Map(); +const texts = new Map(); +const modes = new Map(); +let lostRpc: { path: string; remaining: number } | undefined; +let totalWrites = 0; +const realFetch = globalThis.fetch; +globalThis.fetch = (async (input, init) => { + const url = new URL(String(input)); + if (url.origin === api.origin) { + const response = await realFetch(input, init); + if (lostRpc && url.pathname.endsWith(lostRpc.path) && lostRpc.remaining > 0 && response.ok) { + lostRpc.remaining--; await response.text(); throw new TypeError('injected response loss after database commit'); + } + return response; + } + const token = (new Headers(init?.headers).get('authorization') || '').replace(/^Bearer /, ''); + const owner = tokens.get(token); + assert.ok(owner, 'provider receives only the intended fixture tenant credential'); + if (url.hostname === 'api.x.com' && url.pathname === '/2/users/me') { + if (modes.get(owner) === 'pause') sql(`UPDATE public.user_settings SET automation_publish_enabled = false WHERE user_id = ${literal(owner)}::uuid;`); + await new Promise(resolve => setImmediate(resolve)); + return Response.json({ data: { id: `account-${owner}` } }); + } + if ((url.hostname === 'api.x.com' && url.pathname === '/2/tweets') || (url.hostname === 'api.linkedin.com' && url.pathname === '/v2/ugcPosts')) { + assert.equal(init?.method, 'POST'); totalWrites++; + const providerId = String(900000 + totalWrites); + posts.set(owner, (posts.get(owner) || 0) + 1); + const body = JSON.parse(String(init?.body)); + const text = body.text || body.specificContent?.['com.linkedin.ugc.ShareContent']?.shareCommentary?.text; + texts.set(owner, [...(texts.get(owner) || []), text]); + assert.ok(Number(sql(`SELECT count(*) FROM public.publication_attempts WHERE user_id = ${literal(owner)}::uuid AND state = 'dispatching';`)) > 0); + if (modes.get(owner) === 'timeout') throw new TypeError('injected lost provider response'); + if (modes.get(owner) === 'reject' || url.hostname === 'api.linkedin.com') return Response.json({ message: 'fixture payload rejection' }, { status: 422 }); + if (modes.get(owner) === 'late') { + const queueId = sql(`SELECT queue_item_id FROM public.publication_attempts WHERE user_id = ${literal(owner)}::uuid AND state = 'dispatching';`); + assert.equal((await reconcilePublication({ userId: owner, queueItemId: queueId, platform: 'x' }, Date.now() + 300_000)).outcome, 'unknown'); + } + return Response.json({ data: { id: providerId } }, { status: 201 }); + } + throw new Error(`External network forbidden in integration tests: ${url.origin}`); +}) as typeof fetch; + +function seed(platform: Fixture['row']['platform'] = 'x', existingUser?: string): Fixture { + const user = existingUser || randomUUID(), id = randomUUID(), token = `fixture-${user}`; + tokens.set(token, user); + const scheduled = new Date(Date.now() - 60_000).toISOString(); + if (!existingUser) { + sql(`INSERT INTO auth.users (id, email, raw_user_meta_data) VALUES (${literal(user)}::uuid, ${literal(`${user}@example.invalid`)}, '{}'::jsonb); + UPDATE public.profiles SET subscription_status = 'active' WHERE user_id = ${literal(user)}::uuid; + INSERT INTO public.user_settings (user_id, x_enabled, linkedin_enabled, threads_enabled, instagram_enabled, automation_enabled, automation_publish_enabled) + VALUES (${literal(user)}::uuid, true, true, true, true, true, true) + ON CONFLICT (user_id) DO UPDATE SET x_enabled = true, linkedin_enabled = true, threads_enabled = true, instagram_enabled = true, automation_enabled = true, automation_publish_enabled = true; + INSERT INTO public.user_credentials (user_id, x_oauth2_access_token_enc, linkedin_token_enc, linkedin_person_urn_enc) + VALUES (${literal(user)}::uuid, ${literal(encryptCredential(token))}, ${literal(encryptCredential(token))}, ${literal(encryptCredential(`urn:li:person:${user}`))}) + ON CONFLICT (user_id) DO UPDATE SET x_oauth2_access_token_enc = EXCLUDED.x_oauth2_access_token_enc, linkedin_token_enc = EXCLUDED.linkedin_token_enc, linkedin_person_urn_enc = EXCLUDED.linkedin_person_urn_enc;`); + } + sql(`INSERT INTO public.queue_items (id, user_id, platform, slot_index, scheduled_for, scheduled_local_date, scheduled_timezone, draft_text, source_url, status) + VALUES (${literal(id)}::uuid, ${literal(user)}::uuid, ${literal(platform)}::public.platform, 0, ${literal(scheduled)}::timestamptz, ${literal(scheduled.slice(0, 10))}::date, 'UTC', ${literal(`authorised-${user}`)}, 'https://example.com/reusable-source', 'ready');`); + return { user, token, row: { id, user_id: user, platform, slot_index: 0, scheduled_for: scheduled, draft_text: `authorised-${user}`, status: 'ready' } }; +} +function job(f: Fixture) { + return { id: randomUUID(), user_id: f.user, kind: 'publish_now', payload: { source: 'scheduled', queue_item_id: f.row.id }, status: 'running', created_at: new Date().toISOString() }; +} +const run = (f: Fixture) => runWithRuntimeScope(() => worker.publishQueueRow(job(f), f.row, {}), { userId: f.user }); +const state = (f: Fixture) => loadPublicationStateForQueueItem(f.user, f.row.id); +const target = (f: Fixture) => ({ userId: f.user, queueItemId: f.row.id, platform: f.row.platform }); +let scenarios = 0; +async function test(name: string, fn: () => Promise) { + await fn(); scenarios++; console.log(`PASS ${scenarios}: ${name}`); +} + +async function main() { + await test('real scheduled Worker records exact immutable acceptance and handles duplicate delivery', async () => { + const f = seed(); + sql(`UPDATE public.queue_items SET draft_text = 'approved revision before claim' WHERE id = ${literal(f.row.id)}::uuid;`); + const r = await run(f); + assert.equal(r.outcome, 'accepted'); assert.ok(r.publishHistoryId); + assert.deepEqual(texts.get(f.user), ['approved revision before claim']); + const receipt = await state(f); + assert.equal(receipt.history?.queue_item_id, f.row.id); + assert.equal(receipt.history?.publication_attempt_id, receipt.attempt?.id); + assert.equal(receipt.history?.post_text, 'approved revision before claim'); + assert.equal(receipt.attempt?.provider_account_ref, `account-${f.user}`); + assert.equal((await run(f)).outcome, 'accepted'); assert.equal(posts.get(f.user), 1); + assert.throws(() => sql(`UPDATE public.queue_items SET draft_text = 'late edit' WHERE id = ${literal(f.row.id)}::uuid;`)); + }); + await test('concurrent real Postgres claimers have exactly one provider writer', async () => { + const f = seed(); + await Promise.all(Array.from({ length: 8 }, () => run(f))); + assert.equal(posts.get(f.user), 1); + assert.equal(Number(sql(`SELECT count(*) FROM public.publication_attempts WHERE user_id = ${literal(f.user)}::uuid;`)), 1); + assert.equal((await state(f)).attempt?.state, 'accepted'); + }); + await test('two tenants interleave with their own encrypted credentials and source snapshots', async () => { + const a = seed(), b = seed(); + await Promise.all([run(a), run(b)]); + assert.deepEqual(texts.get(a.user), [`authorised-${a.user}`]); + assert.deepEqual(texts.get(b.user), [`authorised-${b.user}`]); + assert.equal((await state(a)).attempt?.provider_account_ref, `account-${a.user}`); + assert.equal((await state(b)).attempt?.provider_account_ref, `account-${b.user}`); + }); + await test('Postgres history failure rolls back acceptance and blocks repeat publication', async () => { + const f = seed(); + sql(`CREATE OR REPLACE FUNCTION public.fixture_fail_history() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.queue_item_id = ${literal(f.row.id)}::uuid THEN RAISE EXCEPTION 'injected history failure'; END IF; RETURN NEW; END; $$; + CREATE TRIGGER fixture_fail_history BEFORE INSERT ON public.publish_history FOR EACH ROW EXECUTE FUNCTION public.fixture_fail_history();`); + const r = await run(f); + assert.equal(r.outcome, 'unknown'); assert.equal(r.providerAccepted, true); + assert.equal((await state(f)).history, undefined); + assert.equal((await state(f)).attempt?.state, 'unknown'); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'publishing'); + await run(f); assert.equal(posts.get(f.user), 1); + sql('DROP TRIGGER fixture_fail_history ON public.publish_history; DROP FUNCTION public.fixture_fail_history();'); + }); + await test('lost accepted RPC responses reconcile from the committed receipt', async () => { + const f = seed(); lostRpc = { path: '/record_publication_accepted', remaining: 3 }; + assert.equal((await run(f)).outcome, 'accepted'); + assert.equal(lostRpc.remaining, 0); lostRpc = undefined; + assert.equal((await state(f)).attempt?.state, 'accepted'); assert.equal(posts.get(f.user), 1); + }); + await test('lost begin-dispatch responses create no provider call and preserve unknown', async () => { + const f = seed(); lostRpc = { path: '/begin_publication_dispatch', remaining: 3 }; + assert.equal((await run(f)).outcome, 'unknown'); + assert.equal(lostRpc.remaining, 0); lostRpc = undefined; + assert.equal(posts.get(f.user) || 0, 0); assert.equal((await state(f)).attempt?.state, 'unknown'); + await run(f); assert.equal(posts.get(f.user) || 0, 0); + }); + await test('lost claim response reuses its token and fence', async () => { + const f = seed(); lostRpc = { path: '/claim_publication_intent', remaining: 1 }; + assert.equal((await run(f)).outcome, 'accepted'); lostRpc = undefined; + assert.equal((await state(f)).intent?.claim_version, 1); assert.equal(posts.get(f.user), 1); + }); + await test('provider timeout remains unknown through stale recovery and repeated jobs', async () => { + const f = seed(); modes.set(f.user, 'timeout'); + assert.equal((await run(f)).outcome, 'unknown'); + assert.equal((await worker.stalePublishJobResult(job(f), [])).outcome, 'unknown'); + await run(f); assert.equal(posts.get(f.user), 1); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'publishing'); + }); + await test('known rejection records failed projection without inventing history', async () => { + const f = seed(); modes.set(f.user, 'reject'); + assert.equal((await run(f)).outcome, 'rejected'); + assert.equal((await state(f)).attempt?.state, 'rejected'); assert.equal((await state(f)).history, undefined); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'failed'); + await run(f); assert.equal(posts.get(f.user), 1); + }); + await test('pause after identity verification is rechecked before dispatch', async () => { + const f = seed(); modes.set(f.user, 'pause'); + assert.equal((await run(f)).failureCode, 'publish_automation_disabled'); + assert.equal((await state(f)).attempt, undefined); assert.equal(posts.get(f.user) || 0, 0); + }); + await test('disabled Meta causes zero provider, media, claim or revision mutations', async () => { + for (const platform of ['threads', 'instagram'] as const) { + const f = seed(platform); + assert.equal((await run(f)).failureCode, 'legacy_meta_publication_disabled'); + assert.equal((await state(f)).intent, undefined); assert.equal(posts.get(f.user) || 0, 0); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'ready'); + } + }); + await test('publish_all retains independent platform outcomes', async () => { + const f = seed(), other = seed('linkedin', f.user); + const r = await runWithRuntimeScope(() => worker.handlePublishAll({ ...job(f), kind: 'publish_all' }, { userId: f.user, settings: {}, credentials: {}, activePlatforms: ['x', 'linkedin'] })); + assert.equal(r.jobStatus, 'completed_with_errors'); + assert.equal((r.published as unknown[]).length, 1); assert.equal((r.failures as unknown[]).length, 1); + assert.equal((await state(f)).attempt?.state, 'accepted'); assert.equal((await state(other)).attempt?.state, 'rejected'); + }); + await test('legacy ambiguous rows stay quarantined without source-URL guesses', async () => { + const f = seed(); + sql(`UPDATE public.queue_items SET status = 'publishing' WHERE id = ${literal(f.row.id)}::uuid;`); + assert.equal((await reconcilePublication(target(f))).failureCode, 'legacy_publication_requires_reconciliation'); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'publishing'); + assert.equal(posts.get(f.user) || 0, 0); + }); + await test('late provider acceptance resolves a stale unknown attempt with exact positive evidence', async () => { + const f = seed(); modes.set(f.user, 'late'); + assert.equal((await run(f)).outcome, 'accepted'); + const s = await state(f); + assert.equal(s.attempt?.state, 'accepted'); assert.ok(s.attempt?.reconciled_at); + assert.equal(s.attempt?.reconciliation_evidence?.kind, 'exact_attempt_provider_acceptance_response'); + assert.equal(s.history?.publication_attempt_id, s.attempt?.id); + assert.equal(s.attempt?.verified_at, null); + await run(f); assert.equal(posts.get(f.user), 1); + }); + await test('expired pre-dispatch ownership releases through the real fenced RPC', async () => { + const f = seed(); await claimPublicationIntent(f.user, f.row.id, randomUUID(), 120); + assert.equal((await reconcilePublication(target(f), Date.now() + 300_000)).outcome, 'retry_wait'); + assert.equal((await state(f)).intent?.state, 'scheduled'); + assert.equal(sql(`SELECT status FROM public.queue_items WHERE id = ${literal(f.row.id)}::uuid;`), 'ready'); + assert.equal(posts.get(f.user) || 0, 0); + }); + await test('orphan dispatch is recovered without a parent job or any provider write', async () => { + const f = seed(); const intent = await claimPublicationIntent(f.user, f.row.id, randomUUID(), 120); + await beginPublicationDispatch({ userId: f.user, intent, dispatchOperationId: randomUUID(), providerAccountRef: `account-${f.user}`, providerIdempotencySupported: false }); + assert.ok(await recoverStalePublications(Date.now() + 300_000) >= 1); + assert.equal((await state(f)).attempt?.state, 'unknown'); + assert.equal(posts.get(f.user) || 0, 0); + }); + console.log(`PUBLICATION_DATABASE_INTEGRATION_PASS scenarios=${scenarios}; real Supabase/Postgres; provider transport intercepted; live_posts=0`); +} +main().catch(error => { console.error(error); process.exitCode = 1; }).finally(() => { globalThis.fetch = realFetch; }); diff --git a/test/publication-executor.test.ts b/test/publication-executor.test.ts new file mode 100644 index 0000000..7426388 --- /dev/null +++ b/test/publication-executor.test.ts @@ -0,0 +1,234 @@ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import { executePublication, reconcilePublication, PublicationPreflightError } from '../src/publication-executor'; +import { PlatformPublishError } from '../src/platform-errors'; +import type { PublicationIntent, PublicationAttempt, PublicationStateSnapshot } from '../src/publication-ledger'; + +type DB = NonNullable[2]>; +type Hooks = Parameters[1]; +const copy = (value: T): T => structuredClone(value); +const now = () => new Date().toISOString(); + +function fixture(id = 'one') { + const target = { userId: `user-${id}`, queueItemId: `queue-${id}`, platform: 'x' as const }; + let state: PublicationStateSnapshot = {}; + let calls = 0; + let releases = 0; + let begins = 0; + const db: DB = { + assertPublicationLedgerContract: async () => ({ contract: 'publication-ledger-v1', migration: '20260907054000', capabilities: [] }), + loadPublicationStateForQueueItem: async () => copy(state), + claimPublicationIntent: async (_user, _queue, token) => { + const old = state.intent; + if (old && old.state !== 'scheduled' && !(old.state === 'claimed' && Date.parse(old.claim_expires_at || '') <= Date.now())) throw new Error('owned'); + const intent: PublicationIntent = { + id: `intent-${id}`, user_id: target.userId, queue_item_id: target.queueItemId, + platform: 'x', queue_status_before_claim: 'ready', queue_revision_updated_at: now(), + payload: { platform: 'x', text: `authorised-${id}`, source_url: 'https://example.com/reusable-source' }, + state: 'claimed', claim_token: token, claim_version: (old?.claim_version || 0) + 1, + claim_expires_at: new Date(Date.now() + 120_000).toISOString(), created_at: now(), updated_at: now(), + }; + state.intent = intent; + return copy(intent); + }, + releasePublicationClaim: async (_user, claim) => { + assert.equal(state.intent?.state, 'claimed'); + assert.equal(state.intent.claim_token, claim.claim_token); + assert.equal(state.intent.claim_version, claim.claim_version); + releases++; + state.intent.state = 'scheduled'; state.intent.claim_token = null; state.intent.claim_expires_at = null; + return copy(state.intent); + }, + beginPublicationDispatch: async input => { + assert.equal(state.intent?.state, 'claimed'); + assert.equal(state.intent.claim_token, input.intent.claim_token); + assert.equal(state.intent.claim_version, input.intent.claim_version); + assert.ok(Date.parse(state.intent.claim_expires_at || '') > Date.now()); + begins++; + const attempt: PublicationAttempt = { + id: `attempt-${id}`, intent_id: state.intent.id, user_id: target.userId, queue_item_id: target.queueItemId, + platform: 'x', attempt_no: 1, dispatch_operation_id: input.dispatchOperationId!, state: 'dispatching', + provider_account_ref: input.providerAccountRef, provider_idempotency_supported: false, + dispatch_started_at: now(), created_at: now(), updated_at: now(), + }; + state.intent.state = 'dispatching'; state.intent.claim_token = null; state.intent.claim_expires_at = null; + state.attempt = attempt; + return copy(attempt); + }, + recordPublicationAccepted: async input => { + assert.ok(state.attempt && state.intent); + assert.equal(input.attempt.id, state.attempt.id); + assert.equal(input.attempt.dispatch_operation_id, state.attempt.dispatch_operation_id); + if (state.attempt.state === 'unknown') throw new Error('reconciliation required'); + state.attempt.state = 'accepted'; state.intent.state = 'accepted'; + state.attempt.external_post_id = input.externalPostId; + state.attempt.outcome_recorded_at = now(); + state.history = { + id: `history-${id}`, user_id: target.userId, platform: 'x', published_at: now(), + queue_item_id: target.queueItemId, publication_intent_id: state.intent.id, + publication_attempt_id: state.attempt.id, external_post_id: input.externalPostId, + }; + return copy(state.attempt); + }, + recordPublicationRejected: async () => { + assert.ok(state.attempt && state.intent); + assert.equal(state.attempt.state, 'dispatching'); + state.intent.state = 'rejected'; state.attempt.state = 'rejected'; + return copy(state.attempt); + }, + recordPublicationUnknown: async input => { + assert.ok(state.attempt && state.intent); + if (state.attempt.state === 'accepted' || state.attempt.state === 'verified') throw new Error('terminal'); + state.intent.state = 'unknown'; state.attempt.state = 'unknown'; + state.attempt.provider_receipt = input.providerReceipt; + return copy(state.attempt); + }, + }; + const hooks: Hooks = { + prepare: async payload => { + assert.ok(Object.isFrozen(payload)); + assert.equal(payload.text, `authorised-${id}`); + return { providerAccountRef: `account-${id}`, send: async () => { + assert.equal(state.attempt?.state, 'dispatching'); + calls++; + return { externalPostId: `post-${id}` }; + } }; + }, + }; + return { target, db, hooks, state: () => state, calls: () => calls, releases: () => releases, begins: () => begins }; +} + +async function check(name: string, run: () => Promise | void) { + await run(); console.log(`ok - ${name}`); +} + +async function main() { + await check('acceptance is durable before bookkeeping and repeat delivery cannot resend', async () => { + const f = fixture(); + const result = await executePublication(f.target, f.hooks, f.db); + assert.equal(result.outcome, 'accepted'); + assert.equal(result.visibilityVerified, false); + assert.equal(f.state().history?.publication_attempt_id, 'attempt-one'); + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'accepted'); + assert.equal(f.calls(), 1); + }); + await check('interleaved duplicate jobs have one provider writer', async () => { + const f = fixture(); + const results = await Promise.all(Array.from({ length: 8 }, () => executePublication(f.target, f.hooks, f.db))); + assert.equal(results.filter(item => item.outcome === 'accepted').length >= 1, true); + assert.equal(f.calls(), 1); assert.equal(f.begins(), 1); + }); + await check('independent tenants preserve distinct payload and account identity', async () => { + const a = fixture('a'), b = fixture('b'); + await Promise.all([executePublication(a.target, a.hooks, a.db), executePublication(b.target, b.hooks, b.db)]); + assert.equal(a.state().attempt?.provider_account_ref, 'account-a'); + assert.equal(b.state().attempt?.provider_account_ref, 'account-b'); + assert.equal(a.state().attempt?.external_post_id, 'post-a'); + assert.equal(b.state().attempt?.external_post_id, 'post-b'); + }); + await check('post-acceptance ancillary failure preserves accepted truth', async () => { + const f = fixture(); f.hooks.afterAccepted = async () => { throw new Error('bookkeeping'); }; + const r = await executePublication(f.target, f.hooks, f.db); + assert.equal(r.outcome, 'accepted'); assert.equal(r.jobStatus, 'completed_with_errors'); + assert.equal(r.bookkeepingPending, true); + await executePublication(f.target, f.hooks, f.db); assert.equal(f.calls(), 1); + }); + await check('lost acceptance response is reconciled from exact receipt without resend', async () => { + const f = fixture(); const record = f.db.recordPublicationAccepted; + f.db.recordPublicationAccepted = async input => { await record(input); throw new Error('response lost'); }; + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'accepted'); + assert.equal(f.calls(), 1); + }); + await check('failed acceptance transaction retains observed provider ID and blocks retry', async () => { + const f = fixture(); f.db.recordPublicationAccepted = async () => { throw new Error('database unavailable'); }; + const r = await executePublication(f.target, f.hooks, f.db); + assert.equal(r.outcome, 'unknown'); assert.equal(r.providerAccepted, true); + assert.equal(r.externalPostId, 'post-one'); + assert.equal(f.state().attempt?.provider_receipt?.observed_external_post_id, 'post-one'); + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'unknown'); + assert.equal(f.calls(), 1); + }); + await check('provider timeout stays unknown and cannot become retryable failure', async () => { + const f = fixture(); const prepare = f.hooks.prepare; + f.hooks.prepare = async payload => { + const ready = await prepare(payload); + return { ...ready, send: async () => { await ready.send(); throw new Error('lost provider response'); } }; + }; + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'unknown'); + assert.equal((await reconcilePublication(f.target, Date.now() + 300_000, f.db)).outcome, 'unknown'); + await executePublication(f.target, f.hooks, f.db); assert.equal(f.calls(), 1); + }); + await check('missing provider identifier is not a successful receipt', async () => { + const f = fixture(); f.hooks.prepare = async () => ({ providerAccountRef: 'a', send: async () => ({ externalPostId: 'posted' }) }); + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'unknown'); + assert.equal(f.state().history, undefined); + }); + await check('known provider rejection creates no acceptance receipt', async () => { + const f = fixture(); f.hooks.prepare = async () => ({ providerAccountRef: 'a', send: async () => { + throw new PlatformPublishError({ platform: 'x', stage: 'post', status: 422, code: 'payload_rejected', userMessage: 'rejected', nextAction: 'fix' }); + } }); + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'rejected'); + assert.equal(f.state().history, undefined); + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'rejected'); + }); + await check('lost begin-dispatch response makes zero provider calls and never releases the attempt', async () => { + const f = fixture(); const begin = f.db.beginPublicationDispatch; + f.db.beginPublicationDispatch = async input => { await begin(input); throw new Error('lost response'); }; + assert.equal((await executePublication(f.target, f.hooks, f.db)).outcome, 'unknown'); + assert.equal(f.calls(), 0); assert.equal(f.releases(), 0); + assert.equal(f.state().attempt?.state, 'unknown'); + }); + await check('preflight rejection releases only its own fenced pre-dispatch claim', async () => { + const f = fixture(); f.hooks.prepare = async () => { throw new PublicationPreflightError('platform_disabled'); }; + const r = await executePublication(f.target, f.hooks, f.db); + assert.equal(r.failureCode, 'platform_disabled'); assert.equal(f.releases(), 1); + assert.equal(f.begins(), 0); assert.equal(f.calls(), 0); + }); + await check('a lease lost during preflight cannot dispatch', async () => { + const f = fixture(); const prepare = f.hooks.prepare; + f.hooks.prepare = async payload => { const p = await prepare(payload); f.state().intent!.claim_version++; return p; }; + await executePublication(f.target, f.hooks, f.db); assert.equal(f.calls(), 0); + }); + await check('late provider completion after recovery never becomes an automatic retry', async () => { + const f = fixture(); const prepare = f.hooks.prepare; + f.hooks.prepare = async payload => { + const ready = await prepare(payload); + return { ...ready, send: async () => { + const accepted = await ready.send(); + await reconcilePublication(f.target, Date.now() + 300_000, f.db); + return accepted; + } }; + }; + const r = await executePublication(f.target, f.hooks, f.db); + assert.equal(r.outcome, 'unknown'); assert.equal(r.providerAccepted, true); + await executePublication(f.target, f.hooks, f.db); assert.equal(f.calls(), 1); + }); + await check('recovery releases expired pre-dispatch claims but never sends', async () => { + const f = fixture(); await f.db.claimPublicationIntent(f.target.userId, f.target.queueItemId, 'token'); + assert.equal((await reconcilePublication(f.target, Date.now() + 300_000, f.db)).outcome, 'retry_wait'); + assert.equal(f.releases(), 1); assert.equal(f.calls(), 0); + }); + await check('schema or identity failure cannot reach provider code', async () => { + const f = fixture(); f.db.assertPublicationLedgerContract = async () => { throw new Error('missing'); }; + assert.equal((await executePublication(f.target, f.hooks, f.db)).failureCode, 'publication_ledger_schema_unavailable'); + assert.equal(f.calls(), 0); + const g = fixture(); const claim = g.db.claimPublicationIntent; + g.db.claimPublicationIntent = async (...args) => ({ ...await claim(...args), user_id: 'other-tenant' }); + assert.equal((await executePublication(g.target, g.hooks, g.db)).failureCode, 'publication_identity_mismatch'); + assert.equal(g.calls(), 0); assert.equal(g.releases(), 0); + }); + await check('legacy rows are quarantined rather than matched by reusable source URL', async () => { + const f = fixture(); const r = await reconcilePublication(f.target, Date.now(), f.db); + assert.equal(r.failureCode, 'legacy_publication_requires_reconciliation'); assert.equal(f.calls(), 0); + }); + await check('production orchestration calls the executor and exact recovery, not the legacy heuristic', () => { + const worker = readFileSync('src/supabase-worker.ts', 'utf8'); + const publish = worker.slice(worker.indexOf('async function publishQueueRow('), worker.indexOf('async function handlePublishNow(')); + assert.ok(publish.includes('executePublication(')); + assert.ok(!publish.includes("supabaseInsert")); + assert.ok(!publish.includes("status: 'failed'")); + assert.ok(!worker.includes('findPublishHistoryForQueueItem')); + assert.ok(worker.includes('await recoverStalePublications(now.getTime())')); + }); +} +main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/test/publication-ledger.test.ts b/test/publication-ledger.test.ts new file mode 100644 index 0000000..01a97d4 --- /dev/null +++ b/test/publication-ledger.test.ts @@ -0,0 +1,240 @@ +import assert from 'node:assert/strict'; + +import config from '../config'; +import { + PUBLICATION_SCHEMA_CONTRACT, + PUBLICATION_SCHEMA_MIGRATION, + REQUIRED_PUBLICATION_CAPABILITIES, + PublicationLedgerContractError, + assertPublicationLedgerContract, + beginPublicationDispatch, + claimPublicationIntent, + createDispatchOperationId, + createPublicationClaimToken, + recordPublicationAccepted, + recordPublicationUnknown, + releasePublicationClaim, +} from '../src/publication-ledger'; + +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('publication claim and dispatch identities are caller-owned UUIDs', async () => { + assert.match( + createPublicationClaimToken(), + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i + ); + assert.match( + createDispatchOperationId(), + /^[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 probe requires the exact complete publication-ledger-v1 head', async () => { + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: PUBLICATION_SCHEMA_CONTRACT, + migration: PUBLICATION_SCHEMA_MIGRATION, + capabilities: [...REQUIRED_PUBLICATION_CAPABILITIES], + }), { status: 200 })) as typeof fetch; + + const contract = await assertPublicationLedgerContract(); + assert.equal(contract.contract, PUBLICATION_SCHEMA_CONTRACT); + assert.equal(contract.migration, PUBLICATION_SCHEMA_MIGRATION); + + globalThis.fetch = (async () => new Response(JSON.stringify({ + contract: PUBLICATION_SCHEMA_CONTRACT, + migration: '20260907053000', + capabilities: REQUIRED_PUBLICATION_CAPABILITIES.filter( + capability => capability !== 'publication-provenance-snapshot-v1' + ), + }), { status: 200 })) as typeof fetch; + + await assert.rejects( + () => assertPublicationLedgerContract(), + (error: unknown) => error instanceof PublicationLedgerContractError + && error.code === 'publication_ledger_schema_unavailable' + && error.message.includes(PUBLICATION_SCHEMA_MIGRATION) + ); + }); + + await test('claim, release, and dispatch preserve exact fencing identities', async () => { + const bodies: Array<{ url: string; body: Record }> = []; + const claimToken = '61000000-0000-4000-8000-000000000001'; + const dispatchOperationId = '62000000-0000-4000-8000-000000000001'; + let claimVersion = 7; + + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push({ url, body }); + + if (url.endsWith('/rpc/claim_publication_intent')) { + return new Response(JSON.stringify([{ + id: 'intent-1', + user_id: 'user-1', + queue_item_id: 'queue-1', + platform: 'linkedin', + queue_status_before_claim: 'ready', + queue_revision_updated_at: '2026-09-07T00:00:00Z', + payload: { + platform: 'linkedin', + text: 'exact snapshot', + source_url: 'https://example.com/source', + }, + state: 'claimed', + claim_token: claimToken, + claim_version: claimVersion, + claim_expires_at: '2026-09-07T00:02:00Z', + created_at: '2026-09-07T00:00:00Z', + updated_at: '2026-09-07T00:00:00Z', + }]), { status: 200 }); + } + + if (url.endsWith('/rpc/release_publication_claim')) { + return new Response(JSON.stringify([{ + id: 'intent-1', + user_id: 'user-1', + queue_item_id: 'queue-1', + platform: 'linkedin', + queue_status_before_claim: 'ready', + queue_revision_updated_at: '2026-09-07T00:00:00Z', + payload: { platform: 'linkedin', text: 'exact snapshot' }, + state: 'scheduled', + claim_token: null, + claim_version: claimVersion, + claim_expires_at: null, + created_at: '2026-09-07T00:00:00Z', + updated_at: '2026-09-07T00:00:01Z', + }]), { status: 200 }); + } + + if (url.endsWith('/rpc/begin_publication_dispatch')) { + return new Response(JSON.stringify([{ + id: 'attempt-1', + intent_id: 'intent-1', + user_id: 'user-1', + queue_item_id: 'queue-1', + platform: 'linkedin', + attempt_no: 1, + dispatch_operation_id: dispatchOperationId, + state: 'dispatching', + provider_account_ref: 'linkedin:member-1', + provider_idempotency_key: null, + provider_idempotency_supported: false, + dispatch_started_at: '2026-09-07T00:00:02Z', + created_at: '2026-09-07T00:00:02Z', + updated_at: '2026-09-07T00:00:02Z', + }]), { status: 200 }); + } + + throw new Error(`unexpected request: ${url}`); + }) as typeof fetch; + + const intent = await claimPublicationIntent('user-1', 'queue-1', claimToken, 120); + assert.equal(intent.claim_token, claimToken); + assert.equal(intent.claim_version, claimVersion); + assert.equal(intent.payload.text, 'exact snapshot'); + + await releasePublicationClaim('user-1', intent, 'safe pre-dispatch release'); + claimVersion += 1; + const reclaimed = { ...intent, claim_version: claimVersion }; + const attempt = await beginPublicationDispatch({ + userId: 'user-1', + intent: reclaimed, + dispatchOperationId, + providerAccountRef: 'linkedin:member-1', + }); + + assert.equal(attempt.dispatch_operation_id, dispatchOperationId); + assert.equal(bodies[0].body.p_claim_token, claimToken); + assert.equal(bodies[1].body.p_claim_version, 7); + assert.equal(bodies[2].body.p_claim_version, 8); + assert.equal(bodies[2].body.p_dispatch_operation_id, dispatchOperationId); + assert.equal(bodies[2].body.p_provider_idempotency_supported, false); + }); + + await test('accepted and unknown outcome writes carry exact attempt operation identity', async () => { + const bodies: Array<{ url: string; body: Record }> = []; + globalThis.fetch = (async (input, init) => { + const url = String(input); + const body = JSON.parse(String(init?.body || '{}')) as Record; + bodies.push({ url, body }); + const state = url.endsWith('/rpc/record_publication_unknown') ? 'unknown' : 'accepted'; + return new Response(JSON.stringify([{ + id: 'attempt-1', + intent_id: 'intent-1', + user_id: 'user-1', + queue_item_id: 'queue-1', + platform: 'linkedin', + attempt_no: 1, + dispatch_operation_id: '62000000-0000-4000-8000-000000000001', + state, + provider_idempotency_supported: false, + external_post_id: state === 'accepted' ? 'post-1' : null, + dispatch_started_at: '2026-09-07T00:00:02Z', + outcome_recorded_at: '2026-09-07T00:00:03Z', + created_at: '2026-09-07T00:00:02Z', + updated_at: '2026-09-07T00:00:03Z', + }]), { status: 200 }); + }) as typeof fetch; + + const attempt = { + id: 'attempt-1', + dispatch_operation_id: '62000000-0000-4000-8000-000000000001', + }; + await recordPublicationAccepted({ + userId: 'user-1', + attempt, + externalPostId: 'post-1', + providerPublishedAt: '2026-09-07T00:00:03Z', + providerReceipt: { status: 201 }, + }); + await recordPublicationUnknown({ + userId: 'user-1', + attempt, + errorCode: 'network_timeout_after_dispatch', + errorMessage: 'provider outcome could not be observed', + providerReceipt: { network: 'timeout' }, + }); + + assert.equal(bodies[0].body.p_attempt_id, 'attempt-1'); + assert.equal( + bodies[0].body.p_dispatch_operation_id, + '62000000-0000-4000-8000-000000000001' + ); + assert.equal(bodies[0].body.p_external_post_id, 'post-1'); + assert.equal(bodies[1].body.p_error_code, 'network_timeout_after_dispatch'); + assert.equal( + bodies[1].body.p_dispatch_operation_id, + '62000000-0000-4000-8000-000000000001' + ); + }); + } finally { + globalThis.fetch = originalFetch; + config.SUPABASE_URL = originalConfig.url; + config.SUPABASE_SERVICE_ROLE_KEY = originalConfig.role; + config.CREDENTIAL_ENCRYPTION_KEY = originalConfig.encryption; + } +} + +void main(); diff --git a/test/publication-outcome.test.ts b/test/publication-outcome.test.ts new file mode 100644 index 0000000..1d6c973 --- /dev/null +++ b/test/publication-outcome.test.ts @@ -0,0 +1,77 @@ +import assert from 'node:assert/strict'; + +import { HttpError } from '../src/errors'; +import { PlatformPublishError } from '../src/platform-errors'; +import { classifyPostDispatchError } from '../src/publication-outcome'; + +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 { + await test('explicit provider 4xx post response is a known rejection', async () => { + const result = classifyPostDispatchError(new PlatformPublishError({ + platform: 'linkedin', + stage: 'post', + code: 'provider_http_error', + userMessage: 'Provider rejected the post.', + nextAction: 'Fix the request and retry.', + status: 422, + })); + assert.equal(result.outcome, 'rejected'); + assert.equal(result.providerReceipt.http_status, 422); + }); + + await test('408 remains unknown because the external effect may have happened', async () => { + const result = classifyPostDispatchError(new PlatformPublishError({ + platform: 'x', + stage: 'post', + code: 'provider_timeout', + userMessage: 'Provider timed out.', + nextAction: 'Reconcile the exact attempt.', + status: 408, + })); + assert.equal(result.outcome, 'unknown'); + }); + + await test('provider 5xx remains unknown after the dispatch boundary', async () => { + const result = classifyPostDispatchError(new PlatformPublishError({ + platform: 'linkedin', + stage: 'post', + code: 'provider_http_error', + userMessage: 'Provider failed after receiving the request.', + nextAction: 'Reconcile the exact attempt.', + status: 503, + })); + assert.equal(result.outcome, 'unknown'); + }); + + await test('generic HTTP errors are not evidence of provider rejection', async () => { + assert.equal( + classifyPostDispatchError(new HttpError(400, 'bad request', { code: 'UPSTREAM_HTTP_ERROR' })).outcome, + 'unknown' + ); + assert.equal( + classifyPostDispatchError(new HttpError(504, 'timeout', { code: 'UPSTREAM_TIMEOUT' })).outcome, + 'unknown' + ); + assert.equal( + classifyPostDispatchError(new HttpError(502, 'network failed', { code: 'UPSTREAM_REQUEST_FAILED' })).outcome, + 'unknown' + ); + }); + + await test('unclassified post-dispatch exceptions fail conservative to unknown', async () => { + const result = classifyPostDispatchError(new Error('missing provider receipt')); + assert.equal(result.outcome, 'unknown'); + assert.equal(result.providerReceipt.outcome_classification, 'ambiguous_unknown'); + }); +} + +void main(); diff --git a/test/publication-receipts.test.ts b/test/publication-receipts.test.ts new file mode 100644 index 0000000..d5a18e8 --- /dev/null +++ b/test/publication-receipts.test.ts @@ -0,0 +1,47 @@ +import assert from 'node:assert/strict'; +import config from '../config'; +import { recordObservedAcceptance } from '../src/publication-receipts'; + +async function main(): Promise { + const originalFetch = globalThis.fetch; + const previous = { SUPABASE_URL: config.SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY: config.SUPABASE_SERVICE_ROLE_KEY, CREDENTIAL_ENCRYPTION_KEY: config.CREDENTIAL_ENCRYPTION_KEY }; + Object.assign(config, { SUPABASE_URL: 'https://example.supabase.co', SUPABASE_SERVICE_ROLE_KEY: 'fixture-service-role', CREDENTIAL_ENCRYPTION_KEY: 'fixture-key' }); + const input = { userId: 'owner', attempt: { id: 'attempt', dispatch_operation_id: 'dispatch' }, externalPostId: '12345' }; + try { + const paths: string[] = []; + let evidence: Record = {}; + globalThis.fetch = (async (url, init) => { + paths.push(String(url)); + if (String(url).endsWith('/record_publication_accepted')) { + return Response.json({ message: 'publication_unknown_requires_reconciliation', code: 'P0001' }, { status: 400 }); + } + const body = JSON.parse(String(init?.body)); + evidence = body.p_reconciliation_evidence; + assert.equal(body.p_external_post_id, '12345'); + assert.equal(body.p_user_id, 'owner'); + assert.equal(body.p_attempt_id, 'attempt'); + assert.equal(body.p_dispatch_operation_id, 'dispatch'); + return Response.json([{ id: 'attempt', state: 'accepted', external_post_id: '12345' }]); + }) as typeof fetch; + assert.equal((await recordObservedAcceptance(input)).state, 'accepted'); + assert.equal(paths.length, 2); + assert.ok(paths[1].endsWith('/resolve_publication_unknown_accepted')); + assert.equal(evidence.kind, 'exact_attempt_provider_acceptance_response'); + assert.equal(evidence.attempt_id, 'attempt'); + assert.equal(evidence.dispatch_operation_id, 'dispatch'); + console.log('ok - exact delayed provider acceptance resolves its own unknown attempt with positive evidence'); + + paths.length = 0; + globalThis.fetch = (async url => { + paths.push(String(url)); + return Response.json({ message: 'unrelated validation error', code: 'P0001' }, { status: 400 }); + }) as typeof fetch; + await assert.rejects(() => recordObservedAcceptance(input)); + assert.equal(paths.length, 1); + console.log('ok - arbitrary database failures do not trigger an evidence-resolution shortcut'); + } finally { + globalThis.fetch = originalFetch; + Object.assign(config, previous); + } +} +main().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/test/publish-summary.test.ts b/test/publish-summary.test.ts index 8f10c4a..ddd042b 100644 --- a/test/publish-summary.test.ts +++ b/test/publish-summary.test.ts @@ -1,4 +1,5 @@ import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; import { __test__ } from '../src/supabase-worker'; @@ -12,7 +13,7 @@ function test(name: string, fn: () => void): void { } } -test('scheduled publish success includes normalized summary evidence', () => { +test('legacy success summary still preserves exact recorded receipt identifiers', () => { const result = __test__.publishSuccessResult( { id: 'job-1', @@ -47,147 +48,24 @@ test('scheduled publish success includes normalized summary evidence', () => { '2026-05-17T15:00:59.000Z' ) as any; - assert.equal(result.platform, 'threads'); assert.equal(result.queueItemId, 'queue-1'); assert.equal(result.publishHistoryId, 'history-1'); assert.equal(result.externalPostId, 'external-1'); - assert.equal(result.summary.outcome, 'published'); - assert.equal(result.summary.origin, 'scheduled'); - assert.equal(result.summary.scheduler, 'cloudflare_cron'); - assert.equal(result.summary.final_queue_status, 'published'); - assert.equal(result.summary.publish_history_id, 'history-1'); - assert.equal(result.summary.external_post_id, 'external-1'); - assert.match(result.summary.message, /Scheduled post published to threads/); }); -test('reconciled scheduled publish with publish proof becomes published', () => { - const result = __test__.stalePublishResult( - { - id: 'job-reconciled-scheduled', - user_id: 'tenant-1', - kind: 'publish_now', - status: 'running', - created_at: '2026-05-18T04:00:51.000Z', - started_at: '2026-05-18T04:01:48.000Z', - payload: { - source: 'scheduled', - scheduler: 'cloudflare_cron', - due_at: '2026-05-18T04:00:00.000Z', - queue_item_id: 'queue-threads', - }, - }, - { - id: 'queue-threads', - user_id: 'tenant-1', - platform: 'threads', - status: 'published', - slot_index: 0, - scheduled_for: '2026-05-18T04:00:00.000Z', - source_url: 'https://reddit.example/post', - }, - { - id: 'history-threads', - user_id: 'tenant-1', - platform: 'threads', - external_post_id: '18360373546225198', - published_at: '2026-05-18T04:02:02.333Z', - source_url: 'https://reddit.example/post', - }, - [] - ) as any; - - assert.equal(result.summary.outcome, 'published'); - assert.equal(result.summary.platform, 'threads'); - assert.equal(result.summary.queue_item_id, 'queue-threads'); - assert.equal(result.summary.job_id, 'job-reconciled-scheduled'); - assert.equal(result.summary.origin, 'scheduled'); - assert.equal(result.summary.scheduler, 'cloudflare_cron'); - assert.equal(result.summary.final_queue_status, 'published'); - assert.equal(result.summary.publish_history_id, 'history-threads'); - assert.equal(result.summary.external_post_id, '18360373546225198'); - assert.equal(result.summary.completed_at, '2026-05-18T04:02:02.333Z'); - assert.match(result.summary.message, /Scheduled post published to threads/); - assert.equal(result.error, undefined); - assert.equal(result.jobStatus, undefined); -}); - -test('reconciled manual publish with publish proof becomes published', () => { - const result = __test__.stalePublishResult( - { - id: 'job-reconciled-manual', - user_id: 'tenant-1', - kind: 'publish_now', - status: 'running', - created_at: '2026-05-18T12:00:00.000Z', - started_at: '2026-05-18T12:00:01.000Z', - payload: { - queue_item_id: 'queue-linkedin', - }, - }, - { - id: 'queue-linkedin', - user_id: 'tenant-1', - platform: 'linkedin', - status: 'published', - slot_index: 2, - scheduled_for: '2026-05-18T11:00:00.000Z', - source_url: 'https://reddit.example/linkedin', - }, - { - id: 'history-linkedin', - user_id: 'tenant-1', - platform: 'linkedin', - external_post_id: 'urn:li:share:1', - published_at: '2026-05-18T12:00:30.000Z', - source_url: 'https://reddit.example/linkedin', - }, - [] - ) as any; - - assert.equal(result.summary.outcome, 'published'); - assert.equal(result.summary.origin, 'manual'); - assert.equal(result.summary.publish_history_id, 'history-linkedin'); - assert.equal(result.summary.external_post_id, 'urn:li:share:1'); - assert.match(result.summary.message, /Published to linkedin/); +test('production stale publication recovery has no log/status reconciliation fallback', () => { + const worker = readFileSync('src/supabase-worker.ts', 'utf8'); + assert.ok(!worker.includes('function stalePublishResult(')); + assert.ok(!worker.includes('function publishStageStarted(')); + assert.ok(!worker.includes('findPublishHistoryForQueueItem')); + assert.ok(worker.includes( + 'return reconcilePublication({ userId: job.user_id, queueItemId: row.id, platform: row.platform });' + )); + assert.ok(worker.includes('await recoverStalePublications(now.getTime());')); }); -test('unknown publish state remains warning without publish history proof', () => { - const result = __test__.stalePublishResult( - { - id: 'job-unknown', - user_id: 'tenant-1', - kind: 'publish_now', - status: 'running', - created_at: '2026-05-18T12:00:00.000Z', - started_at: '2026-05-18T12:00:01.000Z', - payload: { - source: 'scheduled', - scheduler: 'cloudflare_cron', - queue_item_id: 'queue-x', - }, - }, - { - id: 'queue-x', - user_id: 'tenant-1', - platform: 'x', - status: 'publishing', - slot_index: 2, - scheduled_for: '2026-05-18T11:00:00.000Z', - source_url: 'https://reddit.example/x', - }, - undefined, - [{ - created_at: '2026-05-18T12:00:02.000Z', - level: 'info', - message: 'published_queue_item', - context: { - jobId: 'job-unknown', - queueItemId: 'queue-x', - }, - }] - ) as any; - - assert.equal(result.summary.outcome, 'blocked'); - assert.equal(result.summary.failureCode, 'unknown_publish_state'); - assert.equal(result.jobStatus, 'failed'); +test('unknown publication guidance never authorises a blind retry', () => { + const worker = readFileSync('src/supabase-worker.ts', 'utf8'); + assert.ok(worker.includes('Reconcile publication identity before retrying. Do not recreate the post.')); + assert.ok(!worker.includes('If it is not live, retry this queue item manually.')); });