diff --git a/docs/cloud-native-threads-rollout.md b/docs/cloud-native-threads-rollout.md new file mode 100644 index 0000000..ef0067f --- /dev/null +++ b/docs/cloud-native-threads-rollout.md @@ -0,0 +1,21 @@ +# Cloud-native Threads controlled publishing + +This change is **disabled by default**. It does not authorise general provider dispatch, alter production credentials or publish content by itself. + +## Existing mechanism + +The verified owner job used a Cloudflare controlled tick with a local authenticated Threads relay. The worker has tenant-encrypted credentials (`user_credentials.threads_token_enc`) and scoped token persistence, but its controlled Threads send previously required the relay endpoint. + +## New gated mechanism + +For an exact approved controlled job, `THREADS_CLOUD_NATIVE_ENABLED=true` selects direct Threads Graph API transport. The worker still requires the existing canary tenant, generation-disabled and provider-dispatch-disabled base settings, an exact controlled target, the durable publication attempt and account verification. It uses the decrypted tenant token only inside the tenant runtime. Expired credentials are renewed through existing scoped token persistence, which encrypts the rotated token in the tenant credential row. It verifies the account, creates a text container, publishes and independently reads back the returned post ID. Errors after the publish request remain uncertain; the publication ledger must reconcile rather than automatically resend. + +## Release gates + +1. Confirm the exact tenant's existing encrypted Threads token is present, decryptable and represents the intended Threads account. Do not copy a local plaintext token into Git, logs or Worker vars. +2. Verify tenant token refresh and rotation in a non-publishing controlled check, including token expiry and account mismatch. Confirm the encrypted row updates under the correct tenant. +3. Complete concurrent-admission and crash/restart tests against the durable publication ledger, including an accepted provider publish with lost response. The direct API does not provide a provider-side idempotency key, so uncertain outcomes must never be blindly replayed. +4. Test a single newly approved job, with explicit owner approval, against the real Threads API. Read back the provider ID and compare the exact ledger intent/attempt/history; do not reuse the completed historical post. +5. Promote the verified worker only after independent CI and production readiness. Leave general generation and dispatch disabled. Retire the local relay only after a cloud-native live proof and restart recovery. + +The source-revision migration is separate and must not be pulled into this release accidentally. diff --git a/src/cloudflare-worker.ts b/src/cloudflare-worker.ts index 6de6cee..7ecb6e4 100644 --- a/src/cloudflare-worker.ts +++ b/src/cloudflare-worker.ts @@ -185,6 +185,10 @@ async function executeScheduledTick(env: Env): Promise { config.SUPABASE_WORKER_CANARY_USER_IDS = new Set([target.userId]); config.SUPABASE_PROVIDER_DISPATCH_ENABLED = true; if (target.platform === 'threads') { + if (env.THREADS_CLOUD_NATIVE_ENABLED === 'true') { + if (!env.CREDENTIAL_ENCRYPTION_KEY) throw Error('threads_cloud_encryption_missing'); + return Response.json({ ok: true, stats: await processPendingSupabaseJobs({ ...target, cloudThreads: true }) }); + } if (typeof env.THREADS_RELAY_ENDPOINT !== 'string' || typeof env.THREADS_RELAY_TOKEN !== 'string') { throw Error('controlled_threads_transport_missing'); } diff --git a/src/supabase-worker.ts b/src/supabase-worker.ts index e241ba3..8690387 100644 --- a/src/supabase-worker.ts +++ b/src/supabase-worker.ts @@ -3,6 +3,7 @@ import type { PublicationAttempt } from './publication-ledger'; type ControlledThreadsAttempt = { attemptId: string; dispatchOperationId: string }; type ControlledPublishTarget = ControlledPublish & { sendThreads?: (text: string, attempt: ControlledThreadsAttempt) => Promise; + cloudThreads?: boolean; }; import { managesConnection, connectionCredentialUpdate, loadConnectionSnapshot, withConnectionSnapshot, assertProviderAvailable, assertQueueConnectionReady } from './connection-runtime'; import type { ConnectionSnapshot } from './connection-lifecycle'; @@ -27,6 +28,7 @@ import { import * as workerClaims from './worker-claims'; import { executePublication, reconcilePublication, recoverStalePublications, PublicationPreflightError } from './publication-executor'; import * as threads from './threads'; +import { publishThreadsCloud } from './threads-cloud-publisher'; import * as x from './x'; import { buildDailyInventoryPlan, type DailyInventoryQueueRow } from './daily-inventory-planner'; import { @@ -3615,12 +3617,11 @@ async function sendControlledThreads( if (!attempt || !uuid.test(attempt.id) || !uuid.test(attempt.dispatch_operation_id) || target.platform !== 'threads' || attempt.state !== 'dispatching' || attempt.platform !== 'threads' || attempt.user_id !== target.userId || attempt.queue_item_id !== target.queueId - || attempt.provider_account_ref !== target.accountId || typeof target.sendThreads !== 'function') { + || attempt.provider_account_ref !== target.accountId || (!target.cloudThreads && typeof target.sendThreads !== 'function')) { throw Error('threads_attempt_invalid'); } - return target.sendThreads(text, { - attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id, - }); + if (target.cloudThreads) return publishThreadsCloud({ accountId: target.accountId, accessToken: config.THREADS_ACCESS_TOKEN, text }); + return target.sendThreads!(text, { attemptId: attempt.id, dispatchOperationId: attempt.dispatch_operation_id }); } async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: UserSettingsRow, target?: ControlledPublishTarget): Promise { @@ -3633,13 +3634,13 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U const execution = await executePublication({ userId: job.user_id, queueItemId: row.id, platform: row.platform, }, { - controlledThreadsRelay: target?.platform === 'threads' && typeof target.sendThreads === 'function', + controlledThreadsRelay: target?.platform === 'threads' && (target.cloudThreads === true || typeof target.sendThreads === 'function'), prepare: async payload => { if (!config.SUPABASE_PROVIDER_DISPATCH_ENABLED) { throw new PublicationPreflightError('rollout_provider_dispatch_disabled'); } // Disabled hosted publishers are rejected before token refresh or paid media work. - if ((payload.platform === 'threads' && !(target?.platform === 'threads' && typeof target.sendThreads === 'function')) || payload.platform === 'instagram') { + if ((payload.platform === 'threads' && !(target?.platform === 'threads' && (target.cloudThreads === true || typeof target.sendThreads === 'function'))) || payload.platform === 'instagram') { throw new PublicationPreflightError('legacy_meta_publication_disabled'); } if (payload.platform === 'facebook') throw new PublicationPreflightError('facebook_paused'); @@ -3666,6 +3667,12 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U if (managesConnection(job.user_id)) await markXCredentialVerified(job.user_id, verification); providerAccountRef = verification.accountId; } else if (payload.platform === 'threads') { + // Only the explicit cloud path refreshes tenant credentials. Refresh + // persists the rotated token in the existing encrypted tenant row. + if (target?.cloudThreads) { + try { await threads.verifyCredentials(); } + catch { await threads.prepareAccessTokenForPublish(); } + } const me = await threads.verifyCredentials(); if (me.accountId !== target?.accountId) throw new PublicationPreflightError('controlled_publish_account_changed'); providerAccountRef = me.accountId; @@ -3717,7 +3724,7 @@ async function publishQueueRow(job: AgentJobRow, row: QueueItemRow, _settings: U send: async attempt => { await assertActiveAgentJob(); if (target && Date.now() >= target.expiresAt) throw Error('controlled_publish_expired'); - if (payload.platform === 'threads' && target?.sendThreads) { + if (payload.platform === 'threads' && target && (target.cloudThreads || target.sendThreads)) { const id = await sendControlledThreads(payload.text, target, attempt); if (!/^[0-9]{1,32}$/.test(id)) throw Error('threads_receipt_invalid'); return { externalPostId: id }; diff --git a/src/threads-cloud-publisher.ts b/src/threads-cloud-publisher.ts new file mode 100644 index 0000000..f8795ac --- /dev/null +++ b/src/threads-cloud-publisher.ts @@ -0,0 +1,31 @@ +/** Direct Threads transport. Caller must hold the durable publication claim. */ +export type ThreadsCloudTransport = (url: string, init: RequestInit) => Promise; +export interface ThreadsCloudRequest { accountId: string; accessToken: string; text: string; } +const idPattern = /^[0-9]{1,32}$/; +export async function publishThreadsCloud(request: ThreadsCloudRequest, transport: ThreadsCloudTransport = fetch): Promise { + if (!idPattern.test(request.accountId) || !request.accessToken.trim() || !request.text.trim()) throw Error('threads_cloud_input_invalid'); + const me = await transport('https://graph.threads.net/v1.0/me?fields=id', { headers: { Authorization: `Bearer ${request.accessToken}` }, redirect: 'error' }); + if (!me.ok) throw Error('threads_cloud_account_unverified'); + const identity = await me.json() as { id?: string }; + if (identity.id !== request.accountId) throw Error('threads_cloud_account_mismatch'); + const create = await transport(`https://graph.threads.net/v1.0/${request.accountId}/threads`, { + method: 'POST', redirect: 'error', headers: { Authorization: `Bearer ${request.accessToken}`, 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ media_type: 'TEXT', text: request.text }), + }); + if (!create.ok) throw Error('threads_cloud_container_unconfirmed'); + const container = await create.json() as { id?: string }; + if (!container.id || !idPattern.test(container.id)) throw Error('threads_cloud_container_unconfirmed'); + // An uncertain publish response must remain fenced by the durable attempt ledger. + const published = await transport(`https://graph.threads.net/v1.0/${request.accountId}/threads_publish`, { + method: 'POST', redirect: 'error', headers: { Authorization: `Bearer ${request.accessToken}`, 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ creation_id: container.id }), + }); + if (!published.ok) throw Error('threads_cloud_publication_unconfirmed'); + const result = await published.json() as { id?: string }; + if (!result.id || !idPattern.test(result.id)) throw Error('threads_cloud_publication_unconfirmed'); + const readback = await transport(`https://graph.threads.net/v1.0/${result.id}?fields=id`, { + headers: { Authorization: `Bearer ${request.accessToken}` }, redirect: 'error', + }); + if (!readback.ok || (await readback.json() as { id?: string }).id !== result.id) throw Error('threads_cloud_readback_unconfirmed'); + return result.id; +} diff --git a/test/threads-cloud-publisher.test.ts b/test/threads-cloud-publisher.test.ts new file mode 100644 index 0000000..7875611 --- /dev/null +++ b/test/threads-cloud-publisher.test.ts @@ -0,0 +1,61 @@ +import { test } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { publishThreadsCloud } from '../src/threads-cloud-publisher'; +const input = { accountId: '123', accessToken: 'test-token', text: 'approved text' }; +test('cloud publishing verifies account, creates container, publishes and reads back', async () => { + const calls: string[] = []; + const send = async (url: string) => { + calls.push(url); + const id = calls.length === 1 ? '123' : calls.length === 2 ? '456' : '789'; + return new Response(JSON.stringify({ id }), { status: 200 }); + }; + assert.equal(await publishThreadsCloud(input, send), '789'); + assert.equal(calls.length, 4); +}); +test('wrong account never creates container', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { calls++; return new Response('{"id":"999"}'); }), /account_mismatch/); + assert.equal(calls, 1); +}); +test('missing credentials do not call provider', async () => { + await assert.rejects(publishThreadsCloud({ ...input, accessToken: '' }, async () => { throw Error('unexpected call'); }), /input_invalid/); +}); +test('lost publication response stays unconfirmed, never retries', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + if (calls === 3) throw Error('connection_lost'); + return new Response(JSON.stringify({ id: calls === 1 ? '123' : '456' })); + }), /connection_lost/); + assert.equal(calls, 3); +}); + +test('account verification failure never creates a container', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { calls++; return new Response('{}', { status: 401 }); }), /account_unverified/); + assert.equal(calls, 1); +}); +test('failed container creation never calls publish', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return calls === 1 ? new Response('{"id":"123"}') : new Response('{}', { status: 400 }); + }), /container_unconfirmed/); + assert.equal(calls, 2); +}); +test('provider receipt mismatch never confirms success', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return new Response(JSON.stringify({ id: ['123', '456', '789', '790'][calls - 1] })); + }), /readback_unconfirmed/); + assert.equal(calls, 4); +}); +test('failed publish is fenced without retry', async () => { + let calls = 0; + await assert.rejects(publishThreadsCloud(input, async () => { + calls++; + return calls === 3 ? new Response('{}', { status: 503 }) : new Response(JSON.stringify({ id: calls === 1 ? '123' : '456' })); + }), /publication_unconfirmed/); + assert.equal(calls, 3); +});