From 15c9d0f48c8ad57ca739f3714a5729a320cdeccd Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 22 Sep 2026 20:41:02 +0000 Subject: [PATCH] fix(onboarding): restore secure organization creation --- server/public/onboarding.html | 200 ++- server/src/addie/jobs/job-definitions.ts | 17 + .../organization-onboarding-reconciliation.ts | 55 + ...606_organization_onboarding_operations.sql | 112 ++ server/src/http.ts | 48 +- server/src/routes/member-agents.ts | 6 +- server/src/routes/organizations.ts | 123 +- .../exact-credential-authorization-lock.ts | 28 + server/src/services/organization-bootstrap.ts | 1557 ++++++++++++++++- .../organization-membership-mutation.ts | 13 +- .../integration/claim-prospect-org.test.ts | 4 +- .../implicit-onboarding-containment.test.ts | 72 +- .../integration/join-request-approval.test.ts | 4 +- .../member-agents-auto-bootstrap.test.ts | 4 +- ...rganization-onboarding-restoration.test.ts | 1080 ++++++++++++ ...zations-tier-not-caller-controlled.test.ts | 17 +- 16 files changed, 3098 insertions(+), 242 deletions(-) create mode 100644 server/src/addie/jobs/organization-onboarding-reconciliation.ts create mode 100644 server/src/db/migrations/606_organization_onboarding_operations.sql create mode 100644 server/src/services/exact-credential-authorization-lock.ts create mode 100644 server/tests/integration/organization-onboarding-restoration.test.ts diff --git a/server/public/onboarding.html b/server/public/onboarding.html index 6cbe5c21d45..bc04f1bbeb1 100644 --- a/server/public/onboarding.html +++ b/server/public/onboarding.html @@ -469,14 +469,15 @@

You have pending invitations

@@ -519,7 +520,7 @@

Company Membership

Yes, my company is already a member
- Join your company's existing membership. You may need an admin to approve your access. + Accept an existing invitation, or ask an organization owner to invite your exact signed-in email.
@@ -540,7 +541,7 @@

Company Membership

← Back

Find Your Company

- Search for your company or check if we found a match based on your email. + Search for your company, then ask an owner to invite your exact signed-in email. Self-service join requests are temporarily unavailable.

@@ -575,9 +576,9 @@

- + @@ -891,6 +892,8 @@

Set up your brand

let pendingInvitations = []; let joinableOrgs = []; let userDomain = null; + let currentLegalVersions = null; + const onboardingIdempotency = {}; // Set to true right before re-entering init() after the name-capture form // succeeds. Prevents the gate from re-firing if /api/me momentarily still // returns a null first_name (cache, read-after-write race). @@ -909,6 +912,7 @@

Set up your brand

const data = await response.json(); currentUser = data.user; + await loadLegalVersions(); // Gate: when WorkOS + Slack fallbacks both produced no name, ask the // user before letting them proceed. Credentials, member announcements, @@ -1007,6 +1011,29 @@

Set up your brand

} } + async function loadLegalVersions() { + if (currentLegalVersions) return currentLegalVersions; + const responses = await Promise.all([ + fetch('/api/agreement/current?type=terms_of_service'), + fetch('/api/agreement/current?type=privacy_policy') + ]); + if (!responses[0].ok || !responses[1].ok) { + throw new Error('Current legal agreements are unavailable. Please refresh and try again.'); + } + const values = await Promise.all(responses.map(function(response) { return response.json(); })); + if (!values[0].version || !values[1].version) { + throw new Error('Current legal agreement versions are unavailable.'); + } + currentLegalVersions = { terms: values[0].version, privacy: values[1].version }; + document.querySelectorAll('a[href^="/api/agreement?type=terms_of_service"]').forEach(function(link) { + link.href = '/api/agreement?type=terms_of_service&version=' + encodeURIComponent(currentLegalVersions.terms); + }); + document.querySelectorAll('a[href^="/api/agreement?type=privacy_policy"]').forEach(function(link) { + link.href = '/api/agreement?type=privacy_policy&version=' + encodeURIComponent(currentLegalVersions.privacy); + }); + return currentLegalVersions; + } + function showClaimOrgBannerIfPresent() { const params = new URLSearchParams(window.location.search); const claimOrg = params.get('claim_org'); @@ -1023,29 +1050,6 @@

Set up your brand

nameEl.textContent = claimOrgName || 'your organization'; section.style.display = 'block'; - button.addEventListener('click', async () => { - button.disabled = true; - button.textContent = 'Claiming...'; - errorEl.style.display = 'none'; - try { - const response = await fetch('/api/organizations/' + encodeURIComponent(claimOrg) + '/claim', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - credentials: 'include', - }); - const data = await response.json().catch(() => ({})); - if (!response.ok) { - throw new Error(data.message || 'Claim failed'); - } - window.location.href = '/dashboard'; - } catch (err) { - errorEl.textContent = err.message || 'Failed to claim organization. Please contact support.'; - errorEl.style.display = 'block'; - button.disabled = false; - button.textContent = 'Claim this organization'; - } - }); - skip.addEventListener('click', (e) => { e.preventDefault(); // Strip query params — user opted out, treat the rest of onboarding normally. @@ -1079,7 +1083,10 @@

Set up your brand

'' + '
Invited to join as ' + escapeHtml(role) + '
' + '' + - '' + + '
' + + 'Acceptance is temporarily unavailable. Ask an owner to complete access.
' + + 'Contact support' + + '
' + ''; }).join(''); } @@ -1090,24 +1097,6 @@

Set up your brand

return div.innerHTML.replace(/"/g, '"').replace(/'/g, '''); } - async function acceptInvitation(invitationId) { - try { - const response = await fetch('/api/invitations/' + invitationId + '/accept', { - method: 'POST', - }); - - if (!response.ok) { - const data = await response.json(); - throw new Error(data.message || 'Failed to accept invitation'); - } - - // Success - redirect to dashboard - window.location.href = '/dashboard'; - } catch (error) { - showError(error.message); - } - } - // Step 1: Individual vs Company option selection document.getElementById('optionIndividual').addEventListener('click', function() { selectStep1Option('individual'); @@ -1192,7 +1181,10 @@

Set up your brand

'' + '
Invited to join as ' + escapeHtml(role) + '
' + '' + - '' + + '
' + + 'Acceptance is temporarily unavailable. Ask an owner to complete access.
' + + 'Contact support' + + '
' + ''; }).join(''); } else { @@ -1312,9 +1304,10 @@

Set up your brand

? '
Contact: ' + escapeHtml(org.admin_contact) + '
' : ''; - const buttonHtml = org.request_pending - ? '' - : ''; + const buttonHtml = '
' + + 'Ask an owner to invite your signed-in email.
' + + 'Contact support' + + '
'; return '
' + logoHtml + @@ -1358,43 +1351,6 @@

Set up your brand

container.innerHTML = orgsToRender.map(org => renderOrgCard(org)).join(''); } - async function requestToJoin(orgId, button) { - button.disabled = true; - button.textContent = 'Requesting...'; - - try { - const response = await fetch('/api/join-requests', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ organization_id: orgId }), - }); - - const data = await response.json(); - - if (!response.ok) { - throw new Error(data.message || 'Failed to send request'); - } - - // Update button to show pending state - button.textContent = 'Request Pending'; - button.classList.add('btn-pending'); - - // Update local state - const org = joinableOrgs.find(o => o.organization_id === orgId); - if (org) { - org.request_pending = true; - } - - // Show success message - showInfo('Request sent! An admin will review your request.'); - - } catch (error) { - showError(error.message); - button.disabled = false; - button.textContent = 'Request to Join'; - } - } - function showInfo(message) { const errorDiv = document.getElementById('error'); errorDiv.textContent = message; @@ -1417,6 +1373,42 @@

Set up your brand

errorDiv.style.display = 'block'; } + function getOnboardingIdempotencyKey(scope, body) { + var fingerprint = JSON.stringify(body); + var existing = onboardingIdempotency[scope]; + if (existing && existing.fingerprint === fingerprint) return existing.key; + var storageKey = 'organization-onboarding-idempotency:' + scope; + try { + var stored = JSON.parse(sessionStorage.getItem(storageKey) || 'null'); + if (stored && stored.fingerprint === fingerprint && typeof stored.key === 'string') { + onboardingIdempotency[scope] = stored; + return stored.key; + } + } catch (e) { /* storage disabled or malformed */ } + var key = window.crypto && typeof window.crypto.randomUUID === 'function' + ? window.crypto.randomUUID() + : Date.now().toString(36) + '-' + Math.random().toString(36).slice(2) + '-' + Math.random().toString(36).slice(2); + onboardingIdempotency[scope] = { fingerprint: fingerprint, key: key }; + try { sessionStorage.setItem(storageKey, JSON.stringify(onboardingIdempotency[scope])); } catch (e) { /* storage disabled */ } + return key; + } + + function clearOnboardingIdempotencyKey(scope) { + delete onboardingIdempotency[scope]; + try { sessionStorage.removeItem('organization-onboarding-idempotency:' + scope); } catch (e) { /* storage disabled */ } + } + + function releaseOnboardingIdempotencyKeyAfterTerminalFailure(scope, data) { + var preservedErrors = [ + 'organization_onboarding_in_progress', + 'organization_onboarding_retryable', + 'organization_onboarding_reconciliation_required' + ]; + if (data && typeof data.error === 'string' && !preservedErrors.includes(data.error)) { + clearOnboardingIdempotencyKey(scope); + } + } + // Update company standard pricing based on revenue tier function updateCompanyPricing() { var revenueTier = document.querySelector('input[name="revenueTier"]:checked'); @@ -1523,7 +1515,10 @@

Set up your brand

organization_name: orgName, company_type: companyType.value, revenue_tier: revenueTier.value, - marketing_opt_in: marketingOptIn + marketing_opt_in: marketingOptIn, + agreements_accepted: true, + terms_version: currentLegalVersions.terms, + privacy_version: currentLegalVersions.privacy }; // Stash paid-tier intent in localStorage so the dashboard's // /membership page can pre-select it for Stripe checkout. Tier is @@ -1536,15 +1531,20 @@

Set up your brand

try { var response = await fetch('/api/organizations', { method: 'POST', - headers: { 'Content-Type': 'application/json' }, + headers: { + 'Content-Type': 'application/json', + 'Idempotency-Key': getOnboardingIdempotencyKey('company', body) + }, body: JSON.stringify(body), }); var data = await response.json(); if (!response.ok) { + releaseOnboardingIdempotencyKeyAfterTerminalFailure('company', data); throw new Error(data.message || 'Failed to register company'); } + clearOnboardingIdempotencyKey('company'); // Show brand setup step instead of going directly to dashboard showBrandSetup(orgName, corporateDomain, data.organization?.id || data.id); @@ -1590,7 +1590,10 @@

Set up your brand

var body = { organization_name: workspaceName, is_personal: true, - marketing_opt_in: marketingOptIn + marketing_opt_in: marketingOptIn, + agreements_accepted: true, + terms_version: currentLegalVersions.terms, + privacy_version: currentLegalVersions.privacy }; // Stash paid-tier intent in localStorage so the dashboard's // /membership page can pre-select it for Stripe checkout. Tier is @@ -1602,15 +1605,20 @@

Set up your brand

var response = await fetch('/api/organizations', { method: 'POST', - headers: { 'Content-Type': 'application/json' }, + headers: { + 'Content-Type': 'application/json', + 'Idempotency-Key': getOnboardingIdempotencyKey('personal', body) + }, body: JSON.stringify(body), }); var data = await response.json(); if (!response.ok) { + releaseOnboardingIdempotencyKeyAfterTerminalFailure('personal', data); throw new Error(data.message || 'Failed to create workspace'); } + clearOnboardingIdempotencyKey('personal'); // Keep the new personal workspace as the explicit billing context. // Without this, a user who arrived from a company workspace lands diff --git a/server/src/addie/jobs/job-definitions.ts b/server/src/addie/jobs/job-definitions.ts index bb89a9645da..a98bda28017 100644 --- a/server/src/addie/jobs/job-definitions.ts +++ b/server/src/addie/jobs/job-definitions.ts @@ -68,6 +68,8 @@ import { runOrphanOrgAudit, type OrphanOrgAuditResult } from './orphan-org-audit import { NotificationDatabase } from '../../db/notification-db.js'; import { notifyUser } from '../../notifications/notification-service.js'; import { createLogger } from '../../logger.js'; +import { getAuthorizationEnforcementWorkos } from '../../auth/workos-client.js'; +import { runOrganizationOnboardingReconciliationJob } from './organization-onboarding-reconciliation.js'; const logger = createLogger('job-definitions'); @@ -131,6 +133,20 @@ async function runContentCuratorJob() { * Call this on startup before starting jobs. */ export function registerAllJobs(): void { + jobScheduler.register({ + name: 'organization-onboarding-reconciliation', + description: 'Resume fenced exact-credential organization onboarding operations', + interval: { value: 1, unit: 'minutes' }, + initialDelay: { value: 30, unit: 'seconds' }, + executionTimeoutMs: 3 * 60 * 1000, + runner: () => runOrganizationOnboardingReconciliationJob( + getAuthorizationEnforcementWorkos(), + 2, + 10, + ), + shouldLogResult: (result) => result.attempted > 0 || result.manualQueued > 0, + }); + // Document indexer - indexes Google Docs tracked by committees jobScheduler.register({ name: 'document-indexer', @@ -1012,6 +1028,7 @@ export function registerAllJobs(): void { * Job names for conditional startup (e.g., Moltbook jobs only if API key is set) */ export const JOB_NAMES = { + ORGANIZATION_ONBOARDING_RECONCILIATION: 'organization-onboarding-reconciliation', DOCUMENT_INDEXER: 'document-indexer', SUMMARY_GENERATOR: 'summary-generator', RELATIONSHIP_ORCHESTRATOR: 'relationship-orchestrator', diff --git a/server/src/addie/jobs/organization-onboarding-reconciliation.ts b/server/src/addie/jobs/organization-onboarding-reconciliation.ts new file mode 100644 index 00000000000..30d34244070 --- /dev/null +++ b/server/src/addie/jobs/organization-onboarding-reconciliation.ts @@ -0,0 +1,55 @@ +/** + * Recover safe onboarding operations and durably surface ambiguous ones. + * + * Manual-reconciliation rows must not be replayed automatically because the + * provider/local authority outcome is uncertain. The recurring scan creates a + * deduplicated operator escalation instead, so those rows cannot remain silent. + */ +import type { WorkOS } from '@workos-inc/node'; +import { getAuthorizationEnforcementWorkos } from '../../auth/workos-client.js'; +import { createEscalation } from '../../db/escalation-db.js'; +import { + listManualOrganizationOnboardingOperations, + reconcilePendingOrganizationOnboarding, + type OrganizationOnboardingReconciliationResult, +} from '../../services/organization-bootstrap.js'; + +export type OrganizationOnboardingRecoveryJobResult = + OrganizationOnboardingReconciliationResult & { + manualScanned: number; + manualQueued: number; + }; + +export async function runOrganizationOnboardingReconciliationJob( + workos: WorkOS = getAuthorizationEnforcementWorkos(), + pendingLimit = 2, + manualLimit = 10, +): Promise { + const result: OrganizationOnboardingRecoveryJobResult = { + ...await reconcilePendingOrganizationOnboarding(workos, pendingLimit), + manualScanned: 0, + manualQueued: 0, + }; + const manualOperations = await listManualOrganizationOnboardingOperations(manualLimit); + result.manualScanned = manualOperations.length; + + for (const operation of manualOperations) { + await createEscalation({ + workos_user_id: operation.authenticatedWorkosUserId, + category: 'needs_human_action', + priority: 'high', + summary: `Organization onboarding operation ${operation.id} requires reconciliation`, + addie_context: [ + `Operation ${operation.id} is durably fenced in manual_reconciliation.`, + `Provider organization: ${operation.workosOrganizationId ?? 'not durably recorded'}.`, + `Reason: ${operation.lastErrorCode ?? 'provider/local state divergence'}.`, + 'Do not replay ownership from email, canonical person, or a linked credential.', + 'Compare the exact authenticated WorkOS credential, provider organization and owner membership, and local organization state before resolving.', + ].join(' '), + dedup_key: `organization-onboarding-reconciliation:${operation.id}`, + }); + result.manualQueued += 1; + } + + return result; +} diff --git a/server/src/db/migrations/606_organization_onboarding_operations.sql b/server/src/db/migrations/606_organization_onboarding_operations.sql new file mode 100644 index 00000000000..cccbebd2938 --- /dev/null +++ b/server/src/db/migrations/606_organization_onboarding_operations.sql @@ -0,0 +1,112 @@ +-- Durable exact-credential first-owner onboarding. +-- +-- Provider organization and membership writes cannot share a transaction with +-- PostgreSQL. This ledger makes every provider effect recoverable by a stable +-- external ID/idempotency key and keeps the immutable credential proof that +-- authorized the eventual local commit. + +CREATE TABLE organization_onboarding_operations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + authenticated_workos_user_id VARCHAR(255) NOT NULL, + canonical_workos_user_id VARCHAR(255) NOT NULL, + identity_id UUID NOT NULL, + binding_version TEXT NOT NULL, + authorization_epoch BIGINT NOT NULL, + credential_email VARCHAR(255) NOT NULL, + credential_email_verified BOOLEAN NOT NULL, + + request_fingerprint CHAR(64) NOT NULL, + client_idempotency_key VARCHAR(255), + organization_name VARCHAR(100) NOT NULL, + is_personal BOOLEAN NOT NULL, + company_type VARCHAR(100), + revenue_tier VARCHAR(100), + marketing_opt_in BOOLEAN, + verified_domain VARCHAR(255), + terms_version VARCHAR(50) NOT NULL, + privacy_version VARCHAR(50) NOT NULL, + consent_ip VARCHAR(50), + consent_user_agent TEXT, + + provider_idempotency_key UUID NOT NULL UNIQUE DEFAULT gen_random_uuid(), + provider_external_id VARCHAR(255) NOT NULL UNIQUE, + workos_organization_id VARCHAR(255), + workos_organization_name VARCHAR(255), + workos_membership_id VARCHAR(255), + + status VARCHAR(40) NOT NULL DEFAULT 'pending_provider_organization' + CHECK (status IN ( + 'pending_provider_organization', + 'pending_provider_membership', + 'pending_local_commit', + 'compensating', + 'completed', + 'failed', + 'manual_reconciliation' + )), + lease_token UUID, + lease_expires_at TIMESTAMPTZ, + last_error_code VARCHAR(100), + terminal_outcome JSONB, + completed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + + CHECK ((is_personal AND verified_domain IS NULL) + OR (NOT is_personal AND verified_domain IS NOT NULL)), + CHECK ((status = 'completed' AND workos_organization_id IS NOT NULL + AND workos_membership_id IS NOT NULL AND completed_at IS NOT NULL) + OR status <> 'completed'), + CHECK (status NOT IN ('compensating', 'failed') OR terminal_outcome IS NOT NULL) +); + +CREATE UNIQUE INDEX organization_onboarding_one_active_credential + ON organization_onboarding_operations (authenticated_workos_user_id) + WHERE status NOT IN ('completed', 'failed'); + +-- Linked credentials share org-count and personal-workspace limits. Serializing +-- by identity prevents sibling credentials from passing those checks in +-- parallel while authority remains bound to the exact credential above. +CREATE UNIQUE INDEX organization_onboarding_one_active_identity + ON organization_onboarding_operations (identity_id) + WHERE status NOT IN ('completed', 'failed'); + +CREATE UNIQUE INDEX organization_onboarding_one_active_domain + ON organization_onboarding_operations (verified_domain) + WHERE verified_domain IS NOT NULL AND status NOT IN ('completed', 'failed'); + +CREATE UNIQUE INDEX organization_onboarding_client_idempotency + ON organization_onboarding_operations ( + authenticated_workos_user_id, + client_idempotency_key + ) + WHERE client_idempotency_key IS NOT NULL; + +CREATE INDEX organization_onboarding_retry_lookup + ON organization_onboarding_operations ( + authenticated_workos_user_id, + request_fingerprint, + created_at DESC + ); + +CREATE INDEX organization_onboarding_reconciliation_queue + ON organization_onboarding_operations (updated_at, created_at) + WHERE status IN ( + 'pending_provider_organization', + 'pending_provider_membership', + 'pending_local_commit', + 'compensating' + ); + +CREATE INDEX organization_onboarding_manual_reconciliation_lookup + ON organization_onboarding_operations (updated_at, created_at) + WHERE status = 'manual_reconciliation'; + +COMMENT ON TABLE organization_onboarding_operations IS + 'Durable exact-credential first-owner operations; provider effects are reconciled by stable WorkOS external ID before local authority commits'; +COMMENT ON COLUMN organization_onboarding_operations.binding_version IS + 'PostgreSQL xmin captured from the exact identity_workos_users credential binding'; +COMMENT ON COLUMN organization_onboarding_operations.provider_external_id IS + 'Stable WorkOS organization external ID used to recover ambiguous create responses'; +COMMENT ON COLUMN organization_onboarding_operations.terminal_outcome IS + 'Validated public failure outcome replayed after durable provider compensation'; diff --git a/server/src/http.ts b/server/src/http.ts index f8fdd8887ec..5a4fd4b7ab3 100644 --- a/server/src/http.ts +++ b/server/src/http.ts @@ -9077,45 +9077,13 @@ ${p.category ? `${p.category}\n` : ''}${publishedUrl}< } }); - // POST /api/invitations/:invitationId/accept - Accept an invitation + // POST /api/invitations/:invitationId/accept - Contained until acceptance + // is bound to an exact credential and serialized with membership/audit. this.app.post('/api/invitations/:invitationId/accept', requireAuth, async (req, res) => { - try { - const user = req.user!; - const { invitationId } = req.params; - - // Get the invitation to verify it belongs to this user - const invitation = await workos!.userManagement.getInvitation(invitationId); - - if (invitation.email.toLowerCase() !== user.email.toLowerCase()) { - return res.status(403).json({ - error: 'Access denied', - message: 'This invitation is not for your email address', - }); - } - - if (invitation.state !== 'pending') { - return res.status(400).json({ - error: 'Invalid invitation', - message: 'This invitation has already been accepted or has expired', - }); - } - - // Accept the invitation - this creates the membership - await workos!.userManagement.acceptInvitation(invitationId); - - logger.info({ userId: user.id, invitationId, orgId: invitation.organizationId }, 'User accepted invitation'); - - res.json({ - success: true, - message: 'Invitation accepted successfully', - organization_id: invitation.organizationId, - }); - } catch (error) { - logger.error({ err: error }, 'Accept invitation error:'); - res.status(500).json({ - error: 'Failed to accept invitation', - }); - } + return res.status(403).json({ + error: 'organization_invitation_acceptance_unavailable', + message: 'Self-service invitation acceptance is temporarily unavailable while membership is bound to your exact credential. Ask an organization owner to complete access, or contact support.', + }); }); // GET /api/me/joinable-organizations - Get organizations the user can request to join @@ -9213,8 +9181,8 @@ ${p.category ? `${p.category}\n` : ''}${publishedUrl}< // proof to the canonical user. Restore through the explicit consent flow. this.app.post('/api/join-requests', requireAuth, async (_req, res) => { return res.status(403).json({ - error: 'organization_onboarding_disabled', - message: 'Organization join onboarding is temporarily unavailable.', + error: 'organization_join_onboarding_unavailable', + message: 'Self-service join requests are temporarily unavailable. Ask an organization owner to send a WorkOS invitation, or contact support.', }); }); diff --git a/server/src/routes/member-agents.ts b/server/src/routes/member-agents.ts index 05fbaf6b6fa..f65052d4d42 100644 --- a/server/src/routes/member-agents.ts +++ b/server/src/routes/member-agents.ts @@ -152,9 +152,9 @@ export function createMemberAgentsRouter(config: MemberAgentsRouterConfig): Rout ): Promise<{ orgId: string; orgAutoCreated: boolean } | null> { const apiKey = (req as typeof req & { apiKey?: ValidatedApiKey }).apiKey; if (!apiKey && (typeof req.query.org !== 'string' || !req.query.org)) { - res.status(403).json({ - error: 'organization_onboarding_disabled', - message: 'Select an existing organization explicitly with ?org= to register an agent.', + res.status(400).json({ + error: 'explicit_organization_required', + message: 'Create an organization in onboarding first, then retry with its explicit ?org= identifier.', }); return null; } diff --git a/server/src/routes/organizations.ts b/server/src/routes/organizations.ts index 82547f91f69..b594b17a8dd 100644 --- a/server/src/routes/organizations.ts +++ b/server/src/routes/organizations.ts @@ -789,7 +789,16 @@ export function createOrganizationsRouter(): Router { router.post('/', requireAuth, orgCreationRateLimiter, async (req, res) => { try { const user = req.user!; - const { organization_name, is_personal, company_type, revenue_tier, marketing_opt_in } = req.body; + const { + organization_name, + is_personal, + company_type, + revenue_tier, + marketing_opt_in, + agreements_accepted, + terms_version, + privacy_version, + } = req.body; // `membership_tier` and `corporate_domain` are NOT accepted from caller // input. Tier is owned exclusively by the Stripe webhook (any value @@ -812,37 +821,45 @@ export function createOrganizationsRouter(): Router { const outcome = await performCreateOrganization( { - user: { id: user.id, email: user.email }, + user: { + id: user.id, + authWorkosUserId: user.authWorkosUserId, + email: user.email, + authorizationSnapshot: user.authorizationSnapshot, + impersonator: user.impersonator, + }, + accessToken: req.accessToken, + authorizationHeader: req.headers.authorization, + isApiKey: !!(req as Request & { apiKey?: unknown }).apiKey, + isStaticAdminApiKey: + (req as Request & { isStaticAdminApiKey?: boolean }).isStaticAdminApiKey === true, organization_name, - is_personal: !!is_personal, + is_personal, company_type, revenue_tier, marketing_opt_in, + agreements_accepted, + terms_version, + privacy_version, + clientIdempotencyKey: + typeof req.headers['idempotency-key'] === 'string' + ? req.headers['idempotency-key'] + : undefined, isDevUser: !!(isDevModeEnabled() && getDevUser(req)), requestContext: { ip: req.ip || (req.headers['x-forwarded-for'] as string) || 'unknown', userAgent: (req.headers['user-agent'] as string) || 'unknown', }, }, - { workos, orgDb }, + { workos: AUTH_ENABLED ? getAuthorizationEnforcementWorkos() : null }, ); switch (outcome.kind) { - case 'onboarding_disabled': - return res.status(403).json({ - error: 'organization_onboarding_disabled', - message: 'Organization creation and adoption are temporarily unavailable.', - }); case 'created': return res.json({ success: true, organization: { id: outcome.orgId, name: outcome.name }, - }); - case 'adopted': - return res.status(200).json({ - id: outcome.orgId, - name: outcome.name, - adopted: true, + operation_id: outcome.operationId, }); case 'org_limit_reached': return res.status(400).json({ @@ -874,18 +891,86 @@ export function createOrganizationsRouter(): Router { error: 'Invalid revenue tier', message: `revenue_tier must be one of: ${VALID_REVENUE_TIERS.join(', ')}`, }); + case 'invalid_request': + return res.status(400).json({ + error: 'Invalid request', + message: outcome.message, + }); + case 'agreement_acceptance_required': + return res.status(400).json({ + error: 'agreement_acceptance_required', + message: 'You must accept the current Terms of Use and Privacy Policy before creating an organization.', + }); case 'corporate_email_required': return res.status(400).json({ error: 'Corporate email required', message: 'To register a company, you must be signed in with a corporate email address. Personal email domains (Gmail, Yahoo, etc.) cannot be used for company registration.', }); + case 'verified_corporate_email_required': + return res.status(403).json({ + error: 'verified_corporate_email_required', + message: 'Verify your current corporate email with the identity provider, then sign in again before registering a company.', + }); case 'domain_taken': return res.status(409).json({ error: 'Organization exists', - message: `An organization for ${outcome.domain} already exists: "${outcome.existingOrgName}". Please search for it and request to join instead of creating a new one.`, + message: `An organization for ${outcome.domain} already exists: "${outcome.existingOrgName}". Ask an organization owner to invite your exact signed-in email, or contact support.`, + existing_org_id: outcome.existingOrgId, + existing_org_name: outcome.existingOrgName, + }); + case 'prospect_adoption_unavailable': + return res.status(409).json({ + error: 'organization_adoption_unavailable', + message: `A prospect record for ${outcome.domain} already exists as "${outcome.existingOrgName}". Self-service adoption is not yet available under the new credential-security model. Contact support to verify ownership.`, existing_org_id: outcome.existingOrgId, existing_org_name: outcome.existingOrgName, }); + case 'invalid_credential': + return res.status(401).json({ + error: 'invalid_credential', + message: 'Sign in again with a WorkOS user credential to create an organization.', + }); + case 'impersonation_not_allowed': + return res.status(403).json({ + error: 'impersonation_not_allowed', + message: 'End the impersonation session and sign in as the account owner to accept legal terms and create an organization.', + }); + case 'credential_changed': + return res.status(409).json({ + error: 'onboarding_credential_changed', + message: 'Your verified identity changed during onboarding. Sign in again and submit a new request.', + }); + case 'authorization_unavailable': + return res.status(503).json({ + error: 'authorization_unavailable', + message: 'Your current identity could not be verified. No organization authority was committed; please retry.', + }); + case 'idempotency_mismatch': + return res.status(409).json({ + error: 'idempotency_mismatch', + message: 'This Idempotency-Key was already used for a different organization request.', + }); + case 'onboarding_in_progress': + return res.status(409).json({ + error: 'organization_onboarding_in_progress', + message: 'This organization request is already in progress. Retry the same request shortly.', + retryable: true, + ...(outcome.operationId ? { operation_id: outcome.operationId } : {}), + }); + case 'onboarding_retryable': + return res.status(503).json({ + error: 'organization_onboarding_retryable', + message: 'The provider did not confirm every onboarding step. Retry with the same Idempotency-Key; the existing operation will be reconciled.', + retryable: true, + operation_id: outcome.operationId, + }); + case 'onboarding_reconciliation_required': + return res.status(503).json({ + error: 'organization_onboarding_reconciliation_required', + message: 'The organization request needs support reconciliation. Do not submit a different request.', + retryable: false, + operation_id: outcome.operationId, + }); } } catch (error) { logger.error({ err: error }, 'Create organization error'); @@ -1531,9 +1616,9 @@ export function createOrganizationsRouter(): Router { // First-owner recovery requires a separate explicit, consented operation. router.post('/:orgId/claim', requireAuth, async (_req, res) => { - return res.status(403).json({ - error: 'organization_onboarding_disabled', - message: 'Organization claiming is temporarily unavailable.', + return res.status(409).json({ + error: 'organization_adoption_unavailable', + message: 'Self-service prospect adoption is not yet available under the exact-credential security model. Contact support to verify ownership of this existing organization.', }); }); diff --git a/server/src/services/exact-credential-authorization-lock.ts b/server/src/services/exact-credential-authorization-lock.ts new file mode 100644 index 00000000000..6991e1fb8cd --- /dev/null +++ b/server/src/services/exact-credential-authorization-lock.ts @@ -0,0 +1,28 @@ +import type { PoolClient } from 'pg'; + +/** + * Canonical local-commit fence for exact WorkOS credentials. + * + * Every authorization writer uses the same deterministic row order and fails + * closed on contention. The table lock covers a concurrent platform-ban + * insert, for which there may not yet be a row to lock. + */ +export async function lockExactCredentialAuthorizationRows( + db: Pick, + workosUserIds: string[], +): Promise { + const ids = [...new Set(workosUserIds)].sort(); + await db.query( + 'SELECT workos_user_id FROM users WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', + [ids], + ); + await db.query( + 'SELECT workos_user_id FROM authorization_epochs WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', + [ids], + ); + await db.query( + 'SELECT workos_user_id FROM identity_workos_users WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', + [ids], + ); + await db.query('LOCK TABLE bans IN SHARE MODE NOWAIT'); +} diff --git a/server/src/services/organization-bootstrap.ts b/server/src/services/organization-bootstrap.ts index 2faf3656dc2..0db897b8948 100644 --- a/server/src/services/organization-bootstrap.ts +++ b/server/src/services/organization-bootstrap.ts @@ -1,53 +1,1546 @@ /** - * Organization creation/adoption is suspended until the explicit first-owner - * flow can bind immutable credential proof, current verified domain state, - * consent, and a durable operation to serialized membership/audit/epoch writes. - * Never recover ownership implicitly from email or a canonical person ID. + * Durable exact-credential first-owner onboarding. + * + * WorkOS writes happen outside PostgreSQL transactions. A persisted operation + * supplies a stable provider external ID and idempotency key, while the final + * local transaction atomically commits the organization, exact membership, + * domain proof, audit receipt, legal consent, marketing choice, and epoch. + * Prospect adoption intentionally remains fail-closed: ownership is never + * recovered from an email, domain string, linked identity, or canonical user. */ +import { createHash, randomBytes, randomUUID } from 'node:crypto'; +import { domainToASCII } from 'node:url'; +import type { PoolClient } from 'pg'; import type { WorkOS } from '@workos-inc/node'; -import type { OrganizationDatabase } from '../db/organization-db.js'; +import { parse as parseDomain } from 'tldts'; +import { getPool } from '../db/client.js'; +import { createLogger } from '../logger.js'; +import { COMPANY_TYPE_VALUES } from '../config/company-types.js'; +import { VALID_REVENUE_TIERS } from '../db/organization-db.js'; +import { validateOrganizationName } from '../middleware/validation.js'; +import { getCompanyDomain, isFreeEmailDomain, normalizeEmail } from '../utils/email-domain.js'; +import { isInvalidWorkOSJWTError, verifyWorkOSJWT } from '../auth/workos-jwt.js'; +import { bumpAuthorizationEpochs } from '../db/authorization-epoch-db.js'; +import { invalidateSessionsForUsers } from '../middleware/auth.js'; +import { invalidateMembershipCache } from '../db/org-filters.js'; +import type { AuthorizationSnapshot } from '../db/user-authorization-snapshot-db.js'; +import { assertValidBrandDomain, FREE_EMAIL_PROVIDER_DOMAINS } from './identifier-normalization.js'; +import { lockExactCredentialAuthorizationRows } from './exact-credential-authorization-lock.js'; + +const logger = createLogger('organization-bootstrap'); +// One attempt performs several independently bounded (5s, zero-retry) WorkOS +// calls. Keep the lease longer than their cumulative worst case so a normal +// slow attempt cannot be stolen between provider phases. +const LEASE_SECONDS = 90; +const REPLAY_WINDOW_HOURS = 24; export interface CreateOrgRequest { - user: { id: string; email: string }; - organization_name: string; - is_personal: boolean; - company_type?: string; - revenue_tier?: string; - marketing_opt_in?: boolean; - /** - * Whether the request is from a dev-mode user. The route handler computes - * this via `isDevModeEnabled() && getDevUser(req)`; auto-bootstrap callers - * should pass `false` since dev mode is a dashboard-form concern. - */ + user: { + id: string; + authWorkosUserId?: string; + email: string; + authorizationSnapshot?: AuthorizationSnapshot; + impersonator?: { email: string; reason: string | null }; + }; + accessToken?: string; + authorizationHeader?: string; + isApiKey: boolean; + isStaticAdminApiKey: boolean; + organization_name: unknown; + is_personal: unknown; + company_type?: unknown; + revenue_tier?: unknown; + marketing_opt_in?: unknown; + agreements_accepted?: unknown; + terms_version?: unknown; + privacy_version?: unknown; + clientIdempotencyKey?: string; isDevUser: boolean; - /** - * Request-side context recorded with ToS / privacy acceptance. - */ requestContext: { ip: string; userAgent: string }; } -/** - * Discriminated outcome. The caller maps the kind to an HTTP status. - * - * `created` and `adopted` are the success paths; everything else is a - * domain-level failure that callers may translate into 4xx. - */ export type CreateOrgOutcome = - | { kind: 'onboarding_disabled' } - | { kind: 'created'; orgId: string; name: string } - | { kind: 'adopted'; orgId: string; name: string } + | { kind: 'created'; orgId: string; name: string; operationId: string } | { kind: 'org_limit_reached' } | { kind: 'personal_workspace_exists' } | { kind: 'missing_organization_name' } | { kind: 'invalid_organization_name'; message: string } | { kind: 'invalid_company_type' } | { kind: 'invalid_revenue_tier' } + | { kind: 'invalid_request'; message: string } + | { kind: 'agreement_acceptance_required' } | { kind: 'corporate_email_required' } - | { kind: 'domain_taken'; existingOrgId: string; existingOrgName: string; domain: string }; + | { kind: 'verified_corporate_email_required' } + | { kind: 'domain_taken'; existingOrgId: string; existingOrgName: string; domain: string } + | { kind: 'prospect_adoption_unavailable'; existingOrgId: string; existingOrgName: string; domain: string } + | { kind: 'invalid_credential' } + | { kind: 'impersonation_not_allowed' } + | { kind: 'credential_changed' } + | { kind: 'authorization_unavailable' } + | { kind: 'idempotency_mismatch' } + | { kind: 'onboarding_in_progress'; operationId?: string } + | { kind: 'onboarding_retryable'; operationId: string } + | { kind: 'onboarding_reconciliation_required'; operationId: string }; + +type ValidatedInput = { + organizationName: string; + isPersonal: boolean; + companyType: string | null; + revenueTier: string | null; + marketingOptIn: boolean | null; + clientIdempotencyKey: string | null; + termsVersion: string; + privacyVersion: string; +}; + +type CredentialState = { + credential_exists: boolean; + terminal_marker: boolean; + banned: boolean; + email: string | null; + email_verified: boolean; + identity_id: string | null; + canonical_user_id: string | null; + primary_count: string; + binding_version: string | null; + authorization_epoch: string; +}; + +type ProviderIdentity = { + id: string; + email: string; + emailVerified: boolean; + firstName?: string | null; + lastName?: string | null; +}; + +type OperationStatus = + | 'pending_provider_organization' + | 'pending_provider_membership' + | 'pending_local_commit' + | 'compensating' + | 'completed' + | 'failed' + | 'manual_reconciliation'; + +type OnboardingOperation = { + id: string; + authenticated_workos_user_id: string; + canonical_workos_user_id: string; + identity_id: string; + binding_version: string; + authorization_epoch: string; + credential_email: string; + credential_email_verified: boolean; + request_fingerprint: string; + client_idempotency_key: string | null; + organization_name: string; + is_personal: boolean; + company_type: string | null; + revenue_tier: string | null; + marketing_opt_in: boolean | null; + verified_domain: string | null; + terms_version: string; + privacy_version: string; + consent_ip: string | null; + consent_user_agent: string | null; + provider_idempotency_key: string; + provider_external_id: string; + workos_organization_id: string | null; + workos_organization_name: string | null; + workos_membership_id: string | null; + status: OperationStatus; + lease_token: string | null; + lease_expires_at: Date | null; + last_error_code: string | null; + terminal_outcome: unknown; +}; + +type DomainConflict = { + workos_organization_id: string; + name: string; + prospect_status: string | null; + subscription_status: string | null; +}; + +class FinalizationBusinessError extends Error { + constructor(readonly outcome: CreateOrgOutcome) { + super(outcome.kind); + } +} + +class AuthorityChangedError extends Error {} +class ProviderUnavailableError extends Error {} +class ProviderDivergenceError extends Error {} + +function validateInput(input: CreateOrgRequest): CreateOrgOutcome | ValidatedInput { + if (typeof input.organization_name !== 'string' || input.organization_name.length === 0) { + return { kind: 'missing_organization_name' }; + } + const nameValidation = validateOrganizationName(input.organization_name); + if (!nameValidation.valid) { + return { kind: 'invalid_organization_name', message: nameValidation.error || 'invalid' }; + } + if (input.is_personal !== undefined && typeof input.is_personal !== 'boolean') { + return { kind: 'invalid_request', message: 'is_personal must be a boolean' }; + } + if (input.company_type !== undefined && input.company_type !== null + && (typeof input.company_type !== 'string' + || !COMPANY_TYPE_VALUES.includes(input.company_type as never))) { + return { kind: 'invalid_company_type' }; + } + if (input.revenue_tier !== undefined && input.revenue_tier !== null + && (typeof input.revenue_tier !== 'string' + || !(VALID_REVENUE_TIERS as readonly string[]).includes(input.revenue_tier))) { + return { kind: 'invalid_revenue_tier' }; + } + if (input.marketing_opt_in !== undefined && typeof input.marketing_opt_in !== 'boolean') { + return { kind: 'invalid_request', message: 'marketing_opt_in must be a boolean' }; + } + if (input.agreements_accepted !== true) return { kind: 'agreement_acceptance_required' }; + if (typeof input.terms_version !== 'string' || typeof input.privacy_version !== 'string' + || input.terms_version.length === 0 || input.privacy_version.length === 0 + || input.terms_version.length > 50 || input.privacy_version.length > 50) { + return { kind: 'agreement_acceptance_required' }; + } + const clientIdempotencyKey = input.clientIdempotencyKey?.trim() || null; + if (clientIdempotencyKey && (clientIdempotencyKey.length < 8 || clientIdempotencyKey.length > 255 + || /[^\x21-\x7e]/.test(clientIdempotencyKey))) { + return { kind: 'invalid_request', message: 'Idempotency-Key must be 8-255 visible ASCII characters' }; + } + return { + organizationName: input.organization_name.trim(), + isPersonal: input.is_personal === true, + companyType: typeof input.company_type === 'string' ? input.company_type : null, + revenueTier: typeof input.revenue_tier === 'string' ? input.revenue_tier : null, + marketingOptIn: typeof input.marketing_opt_in === 'boolean' ? input.marketing_opt_in : null, + clientIdempotencyKey, + termsVersion: input.terms_version, + privacyVersion: input.privacy_version, + }; +} + +function requestFingerprint(validated: ValidatedInput, provider: ProviderIdentity): string { + return createHash('sha256').update(JSON.stringify({ + organization_name: validated.organizationName, + is_personal: validated.isPersonal, + company_type: validated.companyType, + revenue_tier: validated.revenueTier, + marketing_opt_in: validated.marketingOptIn, + provider_user_id: provider.id, + provider_email: normalizeEmail(provider.email), + provider_email_verified: provider.emailVerified, + terms_version: validated.termsVersion, + privacy_version: validated.privacyVersion, + })).digest('hex'); +} + +function providerStatus(error: unknown): number | undefined { + const candidate = error as { status?: unknown; statusCode?: unknown } | null; + const value = candidate?.status ?? candidate?.statusCode; + return typeof value === 'number' ? value : undefined; +} + +function providerCode(error: unknown): string | undefined { + const value = (error as { code?: unknown } | null)?.code; + return typeof value === 'string' ? value : undefined; +} + +function isProviderNotFound(error: unknown): boolean { + return providerStatus(error) === 404 || providerCode(error) === 'not_found'; +} + +function isAmbiguousProviderFailure(error: unknown): boolean { + const status = providerStatus(error); + return status === undefined || status === 408 || status === 429 || status >= 500; +} + +function isTransientDatabaseFailure(error: unknown): boolean { + const code = (error as { code?: unknown } | null)?.code; + if (typeof code === 'string' && ( + code.startsWith('08') + || ['40001', '40P01', '53300', '53400', '55P03', '57014', '57P01', '57P02', '57P03'].includes(code) + || ['ECONNREFUSED', 'ECONNRESET', 'EPIPE', 'ETIMEDOUT'].includes(code) + )) return true; + const cause = (error as { cause?: unknown } | null)?.cause; + return cause !== undefined && cause !== error && isTransientDatabaseFailure(cause); +} + +async function verifyPresentedCredential(input: CreateOrgRequest, actorId: string): Promise { + if (input.isApiKey || input.isStaticAdminApiKey || input.isDevUser || !input.accessToken) { + return { kind: 'invalid_credential' }; + } + if (input.authorizationHeader !== undefined) { + const match = /^Bearer[\t ]+([^\t ]+)[\t ]*$/i.exec(input.authorizationHeader); + if (!match || match[1] !== input.accessToken) return { kind: 'invalid_credential' }; + } + try { + const verified = await verifyWorkOSJWT(input.accessToken); + if (verified.isM2M || verified.sub !== actorId) return { kind: 'invalid_credential' }; + if (verified.expiresAt !== undefined && verified.expiresAt <= Date.now() / 1000) { + return { kind: 'invalid_credential' }; + } + } catch (error) { + return isInvalidWorkOSJWTError(error) + ? { kind: 'invalid_credential' } + : { kind: 'authorization_unavailable' }; + } + return null; +} + +async function fetchProviderIdentity(workos: WorkOS, actorId: string): Promise { + try { + const user = await workos.userManagement.getUser(actorId); + if (!user || user.id !== actorId || typeof user.email !== 'string' + || typeof user.emailVerified !== 'boolean') { + throw new ProviderDivergenceError('Provider identity did not match the exact credential'); + } + return { + id: user.id, + email: user.email, + emailVerified: user.emailVerified, + firstName: user.firstName, + lastName: user.lastName, + }; + } catch (error) { + if (error instanceof ProviderDivergenceError) throw error; + if (isProviderNotFound(error)) throw new AuthorityChangedError('Credential no longer exists'); + throw new ProviderUnavailableError('Provider identity unavailable'); + } +} + +async function lockCredentialRows(db: PoolClient, actorId: string): Promise { + await lockExactCredentialAuthorizationRows(db, [actorId]); +} + +async function lockIdentityAuthorizationRows(db: PoolClient, identityId: string): Promise { + const before = await db.query<{ workos_user_id: string }>( + `SELECT workos_user_id FROM identity_workos_users + WHERE identity_id = $1 ORDER BY workos_user_id`, + [identityId], + ); + const ids = before.rows.map(row => row.workos_user_id); + if (ids.length === 0) throw new AuthorityChangedError('Identity has no active credentials'); + await lockExactCredentialAuthorizationRows(db, ids); + // Identity lifecycle writers lock the binding before this identity row. + // NOWAIT preserves that order without waiting while holding sibling rows. + await db.query('SELECT id FROM identities WHERE id = $1 FOR UPDATE NOWAIT', [identityId]); + const after = await db.query<{ workos_user_id: string }>( + `SELECT workos_user_id FROM identity_workos_users + WHERE identity_id = $1 ORDER BY workos_user_id`, + [identityId], + ); + if (after.rows.map(row => row.workos_user_id).join('\0') !== ids.join('\0')) { + throw new AuthorityChangedError('Identity credential set changed'); + } +} + +function currentVerifiedCorporateDomain(email: string): string | null { + const extracted = getCompanyDomain(email); + if (!extracted) return null; + const canonical = domainToASCII(extracted).toLowerCase(); + try { + assertValidBrandDomain(canonical); + const parsed = parseDomain(canonical, { allowPrivateDomains: true }); + if (!parsed.publicSuffix || canonical === parsed.publicSuffix + || isFreeEmailDomain(canonical) + || FREE_EMAIL_PROVIDER_DOMAINS.includes(canonical)) { + return null; + } + return canonical; + } catch { + return null; + } +} + +async function readCredentialState(db: Pick, actorId: string): Promise { + const result = await db.query( + `SELECT (u.workos_user_id IS NOT NULL) AS credential_exists, + EXISTS ( + SELECT 1 FROM registry_audit_log audit + WHERE audit.workos_user_id = $1 + AND audit.action IN ('identity_credential_deleted', 'identity_primary_deletion_quarantined') + ) AS terminal_marker, + EXISTS ( + SELECT 1 FROM bans + WHERE scope = 'platform' + AND (expires_at IS NULL OR expires_at > clock_timestamp()) + AND ban_type = 'user' AND entity_id = $1 + ) AS banned, + u.email, COALESCE(u.email_verified, false) AS email_verified, + binding.identity_id, + primary_binding.workos_user_id AS canonical_user_id, + COALESCE(primary_binding.primary_count, 0)::text AS primary_count, + binding.xmin::text AS binding_version, + COALESCE(epoch.epoch, 0)::text AS authorization_epoch + FROM (VALUES (1)) anchor(value) + LEFT JOIN users u ON u.workos_user_id = $1 + LEFT JOIN identity_workos_users binding ON binding.workos_user_id = u.workos_user_id + LEFT JOIN LATERAL ( + SELECT COUNT(*) AS primary_count, MIN(candidate.workos_user_id) AS workos_user_id + FROM identity_workos_users candidate + WHERE candidate.identity_id = binding.identity_id AND candidate.is_primary + ) primary_binding ON TRUE + LEFT JOIN authorization_epochs epoch ON epoch.workos_user_id = u.workos_user_id`, + [actorId], + ); + if (result.rows.length !== 1) throw new ProviderUnavailableError('Credential state unavailable'); + return result.rows[0]; +} + +function assertLiveCredential( + state: CredentialState, + snapshot: AuthorizationSnapshot, + provider: ProviderIdentity, + actorId: string, +): void { + if (!state.credential_exists || state.terminal_marker || state.banned + || !state.identity_id || !state.canonical_user_id || state.primary_count !== '1' + || !state.binding_version) { + throw new AuthorityChangedError('Credential is no longer active'); + } + if (snapshot.authenticatedUserId !== actorId + || snapshot.canonicalUserId !== state.canonical_user_id + || snapshot.identityId !== state.identity_id + || snapshot.bindingVersion !== state.binding_version + || snapshot.authorizationEpoch !== state.authorization_epoch + || normalizeEmail(snapshot.credential.email ?? '') !== normalizeEmail(state.email ?? '') + || snapshot.credential.emailVerified !== state.email_verified + || provider.id !== actorId + || normalizeEmail(provider.email) !== normalizeEmail(state.email ?? '') + || provider.emailVerified !== state.email_verified) { + throw new AuthorityChangedError('Credential proof changed'); + } +} + +function assertOperationAuthority( + state: CredentialState, + operation: OnboardingOperation, + provider: ProviderIdentity, +): void { + if (!operationAuthorityMatches(state, operation, provider)) { + throw new AuthorityChangedError('Operation authority changed'); + } +} + +function operationAuthorityMatches( + state: CredentialState, + operation: OnboardingOperation, + provider: ProviderIdentity, +): boolean { + return state.credential_exists && !state.terminal_marker && !state.banned + && state.primary_count === '1' + && state.identity_id === operation.identity_id + && state.canonical_user_id === operation.canonical_workos_user_id + && state.binding_version === operation.binding_version + && state.authorization_epoch === operation.authorization_epoch + && normalizeEmail(state.email ?? '') === normalizeEmail(operation.credential_email) + && state.email_verified === operation.credential_email_verified + && provider.id === operation.authenticated_workos_user_id + && normalizeEmail(provider.email) === normalizeEmail(operation.credential_email) + && provider.emailVerified === operation.credential_email_verified; +} + +function domainOutcome(domain: string, existing: DomainConflict): CreateOrgOutcome { + const adoptable = !existing.subscription_status + && (!existing.prospect_status || !['joined', 'declined'].includes(existing.prospect_status)); + return adoptable + ? { + kind: 'prospect_adoption_unavailable', + existingOrgId: existing.workos_organization_id, + existingOrgName: existing.name, + domain, + } + : { + kind: 'domain_taken', + existingOrgId: existing.workos_organization_id, + existingOrgName: existing.name, + domain, + }; +} + +function terminalFailureOutcome(operation: OnboardingOperation): CreateOrgOutcome { + const value = operation.terminal_outcome as Partial | null; + switch (value?.kind) { + case 'org_limit_reached': + case 'personal_workspace_exists': + case 'agreement_acceptance_required': + case 'credential_changed': + case 'authorization_unavailable': + return value as CreateOrgOutcome; + case 'domain_taken': + case 'prospect_adoption_unavailable': + if (typeof (value as { existingOrgId?: unknown }).existingOrgId === 'string' + && typeof (value as { existingOrgName?: unknown }).existingOrgName === 'string' + && typeof (value as { domain?: unknown }).domain === 'string') { + return value as CreateOrgOutcome; + } + break; + } + // Legacy or corrupted terminal rows must stay fail-closed and must never be + // projected as a different successful/retryable operation. + return { kind: 'authorization_unavailable' }; +} + +async function findDomainConflict(db: Pick, domain: string): Promise { + const result = await db.query( + `SELECT o.workos_organization_id, o.name, o.prospect_status, o.subscription_status + FROM organizations o + LEFT JOIN organization_domains od + ON od.workos_organization_id = o.workos_organization_id + WHERE LOWER(od.domain) = LOWER($1) + OR LOWER(o.email_domain) = LOWER($1) + ORDER BY CASE WHEN LOWER(od.domain) = LOWER($1) THEN 0 ELSE 1 END, + o.created_at ASC + LIMIT 1`, + [domain], + ); + return result.rows[0] ?? null; +} + +async function countIdentityOrganizations(db: Pick, identityId: string): Promise { + const result = await db.query<{ count: string }>( + `SELECT COUNT(DISTINCT membership.workos_organization_id)::text AS count + FROM identity_workos_users credential + JOIN organization_memberships membership + ON membership.workos_user_id = credential.workos_user_id + WHERE credential.identity_id = $1`, + [identityId], + ); + return Number(result.rows[0]?.count ?? 0); +} + +async function identityHasPersonalWorkspace(db: Pick, identityId: string): Promise { + const result = await db.query( + `SELECT 1 + FROM identity_workos_users credential + JOIN organization_memberships membership + ON membership.workos_user_id = credential.workos_user_id + JOIN organizations organization + ON organization.workos_organization_id = membership.workos_organization_id + WHERE credential.identity_id = $1 AND organization.is_personal = true + LIMIT 1`, + [identityId], + ); + return result.rowCount === 1; +} + +async function currentAgreementVersions(db: Pick): Promise<{ terms: string; privacy: string } | null> { + const result = await db.query<{ agreement_type: string; version: string }>( + `SELECT DISTINCT ON (agreement_type) agreement_type, version + FROM agreements + WHERE agreement_type IN ('terms_of_service', 'privacy_policy') + ORDER BY agreement_type, effective_date DESC, + string_to_array(version, '.')::int[] DESC`, + ); + const terms = result.rows.find(row => row.agreement_type === 'terms_of_service')?.version; + const privacy = result.rows.find(row => row.agreement_type === 'privacy_policy')?.version; + return terms && privacy ? { terms, privacy } : null; +} + +type PrepareResult = + | { kind: 'execute'; operation: OnboardingOperation; leaseToken: string } + | { kind: 'cleanup'; operation: OnboardingOperation; leaseToken: string } + | { kind: 'outcome'; outcome: CreateOrgOutcome }; + +async function prepareOperation( + input: CreateOrgRequest, + validated: ValidatedInput, + provider: ProviderIdentity, + actorId: string, + snapshot: AuthorizationSnapshot, +): Promise { + const pool = getPool(); + const db = await pool.connect(); + const fingerprint = requestFingerprint(validated, provider); + const verifiedDomain = validated.isPersonal ? null : currentVerifiedCorporateDomain(provider.email); + if (!validated.isPersonal && !verifiedDomain) { + db.release(); + return { kind: 'outcome', outcome: { kind: 'corporate_email_required' } }; + } + if (!validated.isPersonal && !provider.emailVerified) { + db.release(); + return { kind: 'outcome', outcome: { kind: 'verified_corporate_email_required' } }; + } + try { + await db.query('BEGIN'); + await db.query("SET LOCAL lock_timeout = '2s'"); + await db.query("SET LOCAL statement_timeout = '5s'"); + await db.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 606))', [`onboarding-user:${actorId}`]); + // Always take a second scope lock. Corporate onboarding serializes the + // verified domain; personal onboarding uses the exact credential so + // provider-controlled identity data cannot bypass a security lock. + const onboardingScope = verifiedDomain + ? `onboarding-domain:${verifiedDomain}` + : `onboarding-personal:${actorId}`; + await db.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 606))', [onboardingScope]); + await lockCredentialRows(db, actorId); + let state = await readCredentialState(db, actorId); + if (!state.identity_id) throw new AuthorityChangedError('Credential identity changed'); + await lockIdentityAuthorizationRows(db, state.identity_id); + state = await readCredentialState(db, actorId); + assertLiveCredential(state, snapshot, provider, actorId); + + // Run both lookups regardless of whether the caller supplied an + // idempotency key. This keeps the authorization-sensitive replay path + // independent of a user-controlled branch; NULL simply cannot match a + // stored key, and keyed requests are excluded from the fallback query. + const keyed = await db.query( + `SELECT * FROM organization_onboarding_operations + WHERE authenticated_workos_user_id = $1 AND client_idempotency_key = $2 + ORDER BY created_at DESC LIMIT 1 FOR UPDATE`, + [actorId, validated.clientIdempotencyKey ?? null], + ); + const keyedOperation = keyed.rows[0]; + if (keyedOperation) { + if (keyedOperation.request_fingerprint !== fingerprint) { + if (keyedOperation.status === 'manual_reconciliation' || keyedOperation.status === 'compensating') { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'onboarding_reconciliation_required', operationId: keyedOperation.id, + } }; + } + if (keyedOperation.status !== 'completed' && keyedOperation.status !== 'failed' + && !operationAuthorityMatches(state, keyedOperation, provider)) { + if (keyedOperation.lease_token && keyedOperation.lease_expires_at + && new Date(keyedOperation.lease_expires_at).getTime() > Date.now()) { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'onboarding_in_progress', operationId: keyedOperation.id, + } }; + } + const leaseToken = randomUUID(); + const claimed = await db.query( + `UPDATE organization_onboarding_operations + SET lease_token = $2, + lease_expires_at = NOW() + ($3::text || ' seconds')::interval, + updated_at = NOW() + WHERE id = $1 RETURNING *`, + [keyedOperation.id, leaseToken, String(LEASE_SECONDS)], + ); + await db.query('COMMIT'); + return { kind: 'cleanup', operation: claimed.rows[0], leaseToken }; + } + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: { kind: 'idempotency_mismatch' } }; + } + if (keyedOperation.status === 'completed') { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'created', orgId: keyedOperation.workos_organization_id!, + name: keyedOperation.workos_organization_name || keyedOperation.organization_name, + operationId: keyedOperation.id, + } }; + } + if (keyedOperation.status === 'failed') { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: terminalFailureOutcome(keyedOperation) }; + } + } + const replay = await db.query( + `SELECT * FROM organization_onboarding_operations + WHERE authenticated_workos_user_id = $1 + AND request_fingerprint = $2 AND status = 'completed' + AND completed_at > NOW() - ($3::text || ' hours')::interval + AND $4::text IS NULL + ORDER BY completed_at DESC LIMIT 1`, + [actorId, fingerprint, String(REPLAY_WINDOW_HOURS), validated.clientIdempotencyKey ?? null], + ); + const replayOperation = replay.rows[0]; + if (replayOperation) { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'created', orgId: replayOperation.workos_organization_id!, + name: replayOperation.workos_organization_name || replayOperation.organization_name, + operationId: replayOperation.id, + } }; + } + + const activeResult = await db.query( + `SELECT * FROM organization_onboarding_operations + WHERE authenticated_workos_user_id = $1 AND status NOT IN ('completed', 'failed') + FOR UPDATE`, + [actorId], + ); + const active = activeResult.rows[0]; + if (active) { + if (active.status === 'manual_reconciliation') { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'onboarding_reconciliation_required', operationId: active.id, + } }; + } + if (active.status === 'compensating' && active.request_fingerprint !== fingerprint) { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { + kind: 'onboarding_reconciliation_required', operationId: active.id, + } }; + } + // Never let a changed request or credential state steal a live worker's + // lease. That worker may be between the provider write and durable local + // persistence; compensation here could otherwise miss the provider + // object, mark the operation failed, and orphan the later write. + if (active.lease_token && active.lease_expires_at + && new Date(active.lease_expires_at).getTime() > Date.now()) { + await db.query('COMMIT'); + return { kind: 'outcome', outcome: { kind: 'onboarding_in_progress', operationId: active.id } }; + } + const sameAuthority = active.identity_id === state.identity_id + && active.canonical_workos_user_id === state.canonical_user_id + && active.binding_version === state.binding_version + && active.authorization_epoch === state.authorization_epoch + && normalizeEmail(active.credential_email) === normalizeEmail(provider.email) + && active.credential_email_verified === provider.emailVerified; + if (!sameAuthority || active.request_fingerprint !== fingerprint) { + const leaseToken = randomUUID(); + await db.query( + `UPDATE organization_onboarding_operations + SET lease_token = $2, lease_expires_at = NOW() + ($3::text || ' seconds')::interval, + updated_at = NOW() + WHERE id = $1`, + [active.id, leaseToken, String(LEASE_SECONDS)], + ); + await db.query('COMMIT'); + return { kind: 'cleanup', operation: { ...active, lease_token: leaseToken }, leaseToken }; + } + const leaseToken = randomUUID(); + const claimed = await db.query( + `UPDATE organization_onboarding_operations + SET lease_token = $2, lease_expires_at = NOW() + ($3::text || ' seconds')::interval, + updated_at = NOW() + WHERE id = $1 RETURNING *`, + [active.id, leaseToken, String(LEASE_SECONDS)], + ); + await db.query('COMMIT'); + return { kind: 'execute', operation: claimed.rows[0], leaseToken }; + } + + if (await countIdentityOrganizations(db, state.identity_id!) >= 10) { + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: { kind: 'org_limit_reached' } }; + } + if (validated.isPersonal && await identityHasPersonalWorkspace(db, state.identity_id!)) { + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: { kind: 'personal_workspace_exists' } }; + } + if (verifiedDomain) { + const existing = await findDomainConflict(db, verifiedDomain); + if (existing) { + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: domainOutcome(verifiedDomain, existing) }; + } + const reserved = await db.query( + `SELECT 1 FROM organization_onboarding_operations + WHERE verified_domain = $1 AND status NOT IN ('completed', 'failed') + LIMIT 1`, + [verifiedDomain], + ); + if (reserved.rowCount) { + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: { kind: 'onboarding_in_progress' } }; + } + } + await db.query('LOCK TABLE agreements IN SHARE MODE NOWAIT'); + const agreements = await currentAgreementVersions(db); + if (!agreements) throw new ProviderUnavailableError('Current legal agreements unavailable'); + if (agreements.terms !== validated.termsVersion + || agreements.privacy !== validated.privacyVersion) { + await db.query('ROLLBACK'); + return { kind: 'outcome', outcome: { kind: 'agreement_acceptance_required' } }; + } + + const operationId = randomUUID(); + const leaseToken = randomUUID(); + const inserted = await db.query( + `INSERT INTO organization_onboarding_operations ( + id, authenticated_workos_user_id, canonical_workos_user_id, identity_id, + binding_version, authorization_epoch, credential_email, credential_email_verified, + request_fingerprint, client_idempotency_key, organization_name, is_personal, + company_type, revenue_tier, marketing_opt_in, verified_domain, + terms_version, privacy_version, consent_ip, consent_user_agent, + provider_idempotency_key, provider_external_id, lease_token, lease_expires_at + ) VALUES ( + $1, $2, $3, $4, $5, $6::bigint, $7, $8, $9, $10, $11, $12, + $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, + NOW() + ($24::text || ' seconds')::interval + ) RETURNING *`, + [ + operationId, actorId, state.canonical_user_id, state.identity_id, + state.binding_version, state.authorization_epoch, provider.email, provider.emailVerified, + fingerprint, validated.clientIdempotencyKey, validated.organizationName, validated.isPersonal, + validated.companyType, validated.revenueTier, validated.marketingOptIn, verifiedDomain, + validated.termsVersion, validated.privacyVersion, + input.requestContext.ip, input.requestContext.userAgent, + randomUUID(), `adcp_onboarding_${operationId}`, leaseToken, String(LEASE_SECONDS), + ], + ); + await db.query('COMMIT'); + return { kind: 'execute', operation: inserted.rows[0], leaseToken }; + } catch (error) { + await db.query('ROLLBACK').catch(() => {}); + if (error instanceof AuthorityChangedError) return { kind: 'outcome', outcome: { kind: 'credential_changed' } }; + if ((error as { code?: string }).code === '23505') { + return { kind: 'outcome', outcome: { kind: 'onboarding_in_progress' } }; + } + logger.error({ error, actorId }, 'Failed to prepare organization onboarding operation'); + return { kind: 'outcome', outcome: { kind: 'authorization_unavailable' } }; + } finally { + db.release(); + } +} + +async function persistOperation( + operationId: string, + leaseToken: string, + patch: { + status?: OperationStatus; + workosOrganizationId?: string; + workosOrganizationName?: string; + workosMembershipId?: string; + errorCode?: string | null; + releaseLease?: boolean; + }, +): Promise { + let result; + try { + result = await getPool().query( + `UPDATE organization_onboarding_operations + SET status = COALESCE($3, status), + workos_organization_id = COALESCE($4, workos_organization_id), + workos_organization_name = COALESCE($5, workos_organization_name), + workos_membership_id = COALESCE($6, workos_membership_id), + last_error_code = $7, + lease_token = CASE WHEN $8 THEN NULL ELSE lease_token END, + lease_expires_at = CASE WHEN $8 THEN NULL ELSE lease_expires_at END, + updated_at = NOW() + WHERE id = $1 AND lease_token = $2 + RETURNING *`, + [ + operationId, leaseToken, patch.status ?? null, + patch.workosOrganizationId ?? null, patch.workosOrganizationName ?? null, + patch.workosMembershipId ?? null, patch.errorCode ?? null, + patch.releaseLease === true, + ], + ); + } catch { + // UPDATE acknowledgement can be lost after PostgreSQL commits. Never + // compensate a provider object on an ambiguous ledger write: a retry can + // reconcile both the pre- and post-commit database states safely. + throw new ProviderUnavailableError('Onboarding operation persistence unavailable'); + } + if (result.rowCount !== 1) throw new ProviderDivergenceError('Operation lease changed'); + return result.rows[0]; +} + +async function renewOperationLease(operationId: string, leaseToken: string): Promise { + let result; + try { + result = await getPool().query( + `UPDATE organization_onboarding_operations + SET lease_expires_at = NOW() + ($3::text || ' seconds')::interval, + updated_at = NOW() + WHERE id = $1 AND lease_token = $2 + AND status NOT IN ('completed', 'failed', 'manual_reconciliation')`, + [operationId, leaseToken, String(LEASE_SECONDS)], + ); + } catch { + throw new ProviderUnavailableError('Onboarding lease renewal unavailable'); + } + if (result.rowCount !== 1) throw new ProviderDivergenceError('Operation lease changed'); +} + +async function liveOperationAuthority( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, +): Promise { + await renewOperationLease(operation.id, leaseToken); + const provider = await fetchProviderIdentity(workos, operation.authenticated_workos_user_id); + const state = await readCredentialState(getPool(), operation.authenticated_workos_user_id); + assertOperationAuthority(state, operation, provider); + return provider; +} + +function validateProviderOrganization(value: unknown): { id: string; name: string } { + const org = value as { id?: unknown; name?: unknown } | null; + if (!org || typeof org.id !== 'string' || typeof org.name !== 'string') { + throw new ProviderDivergenceError('Invalid provider organization response'); + } + return { id: org.id, name: org.name }; +} + +async function ensureProviderOrganization( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, +): Promise { + if (operation.workos_organization_id) { + await renewOperationLease(operation.id, leaseToken); + try { + const providerOrganization = validateProviderOrganization( + await workos.organizations.getOrganizationByExternalId(operation.provider_external_id), + ); + if (providerOrganization.id !== operation.workos_organization_id) { + throw new ProviderDivergenceError('Stored provider organization no longer matches its external ID'); + } + return operation; + } catch (error) { + if (error instanceof ProviderDivergenceError || isProviderNotFound(error)) { + throw new ProviderDivergenceError('Stored provider organization is no longer authoritative'); + } + throw new ProviderUnavailableError('Stored provider organization could not be confirmed'); + } + } + await liveOperationAuthority(workos, operation, leaseToken); + let providerOrganization: { id: string; name: string } | null = null; + try { + providerOrganization = validateProviderOrganization( + await workos.organizations.getOrganizationByExternalId(operation.provider_external_id), + ); + } catch (error) { + if (!isProviderNotFound(error)) { + if (error instanceof ProviderDivergenceError) throw error; + throw new ProviderUnavailableError('Provider organization lookup unavailable'); + } + } + if (!providerOrganization) { + await renewOperationLease(operation.id, leaseToken); + try { + providerOrganization = validateProviderOrganization( + await workos.organizations.createOrganization( + { + name: operation.organization_name, + externalId: operation.provider_external_id, + metadata: { onboarding_operation_id: operation.id }, + }, + { idempotencyKey: operation.provider_idempotency_key }, + ), + ); + } catch (error) { + if (providerStatus(error) === 409) { + try { + providerOrganization = validateProviderOrganization( + await workos.organizations.getOrganizationByExternalId(operation.provider_external_id), + ); + } catch { + throw new ProviderUnavailableError('Provider organization conflict could not be reconciled'); + } + } else if (isAmbiguousProviderFailure(error)) { + throw new ProviderUnavailableError('Provider organization creation was not confirmed'); + } else { + throw error; + } + } + } + operation = await persistOperation(operation.id, leaseToken, { + status: 'pending_provider_membership', + workosOrganizationId: providerOrganization.id, + workosOrganizationName: providerOrganization.name, + }); + await liveOperationAuthority(workos, operation, leaseToken); + return operation; +} + +type ProviderMembership = { + id: string; + userId: string; + organizationId: string; + status: string; + role?: { slug?: string } | null; +}; + +async function readExactProviderMembership(workos: WorkOS, operation: OnboardingOperation): Promise { + let page; + try { + page = await workos.userManagement.listOrganizationMemberships({ + userId: operation.authenticated_workos_user_id, + organizationId: operation.workos_organization_id!, + statuses: ['active', 'inactive', 'pending'], + }); + } catch { + throw new ProviderUnavailableError('Provider membership lookup unavailable'); + } + if (!page || !Array.isArray(page.data)) throw new ProviderDivergenceError('Invalid provider membership response'); + const rows = (page.data as ProviderMembership[]).filter(row => row + && row.userId === operation.authenticated_workos_user_id + && row.organizationId === operation.workos_organization_id); + if (rows.length > 1) throw new ProviderDivergenceError('Provider membership state is ambiguous'); + return rows[0] ?? null; +} + +function assertFirstOwnerMembership(operation: OnboardingOperation, membership: ProviderMembership): void { + if (membership.userId !== operation.authenticated_workos_user_id + || membership.organizationId !== operation.workos_organization_id + || membership.status !== 'active' || membership.role?.slug !== 'owner' + || typeof membership.id !== 'string') { + throw new ProviderDivergenceError('Provider did not confirm the exact first owner'); + } +} + +async function ensureProviderMembership( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, +): Promise { + await liveOperationAuthority(workos, operation, leaseToken); + let membership = await readExactProviderMembership(workos, operation); + if (!membership) { + await renewOperationLease(operation.id, leaseToken); + try { + membership = await workos.userManagement.createOrganizationMembership({ + userId: operation.authenticated_workos_user_id, + organizationId: operation.workos_organization_id!, + roleSlug: 'owner', + }) as ProviderMembership; + } catch (error) { + if (providerCode(error) === 'organization_membership_already_exists' + || providerStatus(error) === 409) { + membership = await readExactProviderMembership(workos, operation); + } else if (isAmbiguousProviderFailure(error)) { + throw new ProviderUnavailableError('Provider owner creation was not confirmed'); + } else { + throw error; + } + } + } + if (!membership) throw new ProviderUnavailableError('Provider owner creation was not confirmed'); + assertFirstOwnerMembership(operation, membership); + operation = await persistOperation(operation.id, leaseToken, { + status: 'pending_local_commit', workosMembershipId: membership.id, + }); + await liveOperationAuthority(workos, operation, leaseToken); + return operation; +} + +async function writeMarketingChoice(db: PoolClient, operation: OnboardingOperation): Promise { + if (operation.marketing_opt_in === null) return; + const token = randomBytes(32).toString('hex'); + const inserted = await db.query<{ id: string; marketing_opt_in: boolean | null }>( + `INSERT INTO user_email_preferences (workos_user_id, email, unsubscribe_token) + VALUES ($1, $2, $3) + ON CONFLICT (workos_user_id) DO UPDATE SET email = EXCLUDED.email + RETURNING id, marketing_opt_in`, + [operation.authenticated_workos_user_id, operation.credential_email, token], + ); + const preference = inserted.rows[0]; + if (preference.marketing_opt_in !== null) return; + await db.query( + `UPDATE user_email_preferences + SET marketing_opt_in = $2, marketing_opt_in_at = NOW(), updated_at = NOW() + WHERE id = $1 AND marketing_opt_in IS NULL`, + [preference.id, operation.marketing_opt_in], + ); + if (operation.marketing_opt_in) { + await db.query('DELETE FROM user_email_category_preferences WHERE user_preference_id = $1', [preference.id]); + } else { + await db.query( + `INSERT INTO user_email_category_preferences (user_preference_id, category_id, enabled) + SELECT $1, id, false FROM email_categories + ON CONFLICT (user_preference_id, category_id) + DO UPDATE SET enabled = false, updated_at = NOW()`, + [preference.id], + ); + } +} + +async function finalizeLocalOperation( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, +): Promise { + const db = await getPool().connect(); + let commitAttempted = false; + try { + await db.query('BEGIN'); + await db.query("SET LOCAL lock_timeout = '2s'"); + await db.query("SET LOCAL statement_timeout = '8s'"); + const locked = await db.query( + 'SELECT * FROM organization_onboarding_operations WHERE id = $1 FOR UPDATE', + [operation.id], + ); + operation = locked.rows[0]; + if (!operation || operation.lease_token !== leaseToken) { + throw new ProviderDivergenceError('Operation lease changed'); + } + if (operation.status === 'completed') { + await db.query('COMMIT'); + return { + kind: 'created', orgId: operation.workos_organization_id!, + name: operation.workos_organization_name || operation.organization_name, + operationId: operation.id, + }; + } + await lockCredentialRows(db, operation.authenticated_workos_user_id); + let state = await readCredentialState(db, operation.authenticated_workos_user_id); + if (state.identity_id !== operation.identity_id) { + throw new AuthorityChangedError('Credential identity changed'); + } + await lockIdentityAuthorizationRows(db, operation.identity_id); + state = await readCredentialState(db, operation.authenticated_workos_user_id); + + // These are the only bounded network reads made while local authorization + // rows are locked. The enforcement WorkOS client has a 5s timeout and no + // retries, so credential/domain state cannot drift across the local commit. + const provider = await fetchProviderIdentity(workos, operation.authenticated_workos_user_id); + assertOperationAuthority(state, operation, provider); + const membership = await readExactProviderMembership(workos, operation); + if (!membership) throw new ProviderDivergenceError('Provider owner disappeared before commit'); + assertFirstOwnerMembership(operation, membership); + if (membership.id !== operation.workos_membership_id) { + throw new ProviderDivergenceError('Provider owner changed before commit'); + } + + await db.query('LOCK TABLE agreements IN SHARE MODE NOWAIT'); + const agreements = await currentAgreementVersions(db); + if (!agreements || agreements.terms !== operation.terms_version + || agreements.privacy !== operation.privacy_version) { + throw new FinalizationBusinessError({ kind: 'agreement_acceptance_required' }); + } + if (await countIdentityOrganizations(db, operation.identity_id) >= 10) { + throw new FinalizationBusinessError({ kind: 'org_limit_reached' }); + } + if (operation.is_personal && await identityHasPersonalWorkspace(db, operation.identity_id)) { + throw new FinalizationBusinessError({ kind: 'personal_workspace_exists' }); + } + if (operation.verified_domain) { + const existing = await findDomainConflict(db, operation.verified_domain); + if (existing) throw new FinalizationBusinessError(domainOutcome(operation.verified_domain, existing)); + } + + await db.query( + `INSERT INTO organizations + (workos_organization_id, name, is_personal, company_type, revenue_tier, email_domain) + VALUES ($1, $2, $3, $4, $5, $6)`, + [ + operation.workos_organization_id, operation.workos_organization_name || operation.organization_name, + operation.is_personal, operation.company_type, operation.revenue_tier, + operation.verified_domain, + ], + ); + await db.query( + `INSERT INTO organization_memberships + (workos_user_id, workos_organization_id, workos_membership_id, email, + first_name, last_name, role, seat_type, provisioning_source, synced_at) + VALUES ($1, $2, $3, $4, $5, $6, 'owner', 'contributor', 'webhook', NOW())`, + [ + operation.authenticated_workos_user_id, operation.workos_organization_id, + operation.workos_membership_id, provider.email, + provider.firstName?.trim() || null, provider.lastName?.trim() || null, + ], + ); + if (operation.verified_domain) { + await db.query( + `INSERT INTO organization_domains + (workos_organization_id, domain, is_primary, verified, source) + VALUES ($1, $2, true, true, 'email_verification')`, + [operation.workos_organization_id, operation.verified_domain], + ); + } + await db.query( + `INSERT INTO registry_audit_log + (workos_organization_id, workos_user_id, action, resource_type, resource_id, details) + VALUES ($1::varchar, $2, 'organization_created', 'organization', $1::varchar, $3)`, + [operation.workos_organization_id, operation.authenticated_workos_user_id, JSON.stringify({ + operation_id: operation.id, + authenticated_workos_user_id: operation.authenticated_workos_user_id, + canonical_workos_user_id: operation.canonical_workos_user_id, + identity_id: operation.identity_id, + binding_version: operation.binding_version, + authorization_epoch: operation.authorization_epoch, + authority: 'exact_credential_first_owner', + provider_identity_verified: true, + verified_domain: operation.verified_domain, + is_personal: operation.is_personal, + company_type: operation.company_type, + revenue_tier: operation.revenue_tier, + })], + ); + for (const [agreementType, version] of [ + ['terms_of_service', operation.terms_version], + ['privacy_policy', operation.privacy_version], + ] as const) { + const accepted = await db.query( + `INSERT INTO user_agreement_acceptances + (workos_user_id, email, agreement_type, agreement_version, ip_address, + user_agent, workos_organization_id) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (workos_user_id, agreement_type, agreement_version) DO NOTHING + RETURNING id`, + [ + operation.authenticated_workos_user_id, provider.email, agreementType, version, + operation.consent_ip, operation.consent_user_agent, operation.workos_organization_id, + ], + ); + if (accepted.rowCount !== 1) { + const existing = await db.query( + `SELECT 1 FROM user_agreement_acceptances + WHERE workos_user_id = $1 AND agreement_type = $2 AND agreement_version = $3`, + [operation.authenticated_workos_user_id, agreementType, version], + ); + if (existing.rowCount !== 1) throw new Error('Agreement acceptance was not recorded'); + } + } + await writeMarketingChoice(db, operation); + const bumped = await bumpAuthorizationEpochs(db, [operation.authenticated_workos_user_id]); + if (bumped.length !== 1) throw new Error('Authorization epoch was not advanced'); + const completed = await db.query( + `UPDATE organization_onboarding_operations + SET status = 'completed', completed_at = NOW(), lease_token = NULL, + lease_expires_at = NULL, last_error_code = NULL, updated_at = NOW() + WHERE id = $1 AND lease_token = $2 AND status = 'pending_local_commit'`, + [operation.id, leaseToken], + ); + if (completed.rowCount !== 1) throw new Error('Onboarding operation state changed'); + commitAttempted = true; + await db.query('COMMIT'); + invalidateMembershipCache(operation.workos_organization_id!); + invalidateSessionsForUsers([operation.authenticated_workos_user_id]); + return { + kind: 'created', orgId: operation.workos_organization_id!, + name: operation.workos_organization_name || operation.organization_name, + operationId: operation.id, + }; + } catch (error) { + await db.query('ROLLBACK').catch(() => {}); + if (commitAttempted) { + const recovered = await getPool().query( + 'SELECT * FROM organization_onboarding_operations WHERE id = $1', [operation.id], + ).catch(() => ({ rows: [] as OnboardingOperation[] })); + const row = recovered.rows[0]; + if (row?.status === 'completed') { + return { + kind: 'created', orgId: row.workos_organization_id!, + name: row.workos_organization_name || row.organization_name, + operationId: row.id, + }; + } + throw new ProviderUnavailableError('Local commit outcome is not yet confirmed'); + } + throw error; + } finally { + db.release(); + } +} + +async function markOperation( + operationId: string, + leaseToken: string, + status: OperationStatus, + errorCode: string, +): Promise { + const result = await getPool().query( + `UPDATE organization_onboarding_operations + SET status = $3, last_error_code = $4, lease_token = NULL, + lease_expires_at = NULL, updated_at = NOW() + WHERE id = $1 AND lease_token = $2`, + [operationId, leaseToken, status, errorCode], + ); + return result.rowCount === 1; +} + +async function beginCompensation( + operation: OnboardingOperation, + leaseToken: string, + reason: string, + outcome: CreateOrgOutcome, +): Promise { + let result; + try { + result = await getPool().query( + `UPDATE organization_onboarding_operations + SET status = 'compensating', last_error_code = $3, + terminal_outcome = $4::jsonb, + lease_expires_at = NOW() + ($5::text || ' seconds')::interval, + updated_at = NOW() + WHERE id = $1 AND lease_token = $2 + AND status NOT IN ('completed', 'failed', 'manual_reconciliation') + RETURNING *`, + [operation.id, leaseToken, reason, JSON.stringify(outcome), String(LEASE_SECONDS)], + ); + } catch { + // The acknowledgement may have been lost after commit. No provider + // deletion is safe until a later worker observes durable compensation. + throw new ProviderUnavailableError('Compensation intent persistence unavailable'); + } + if (result.rowCount !== 1) throw new ProviderDivergenceError('Operation lease changed'); + return result.rows[0]; +} + +async function compensateProviderOrganization( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, + reason: string, + outcome: CreateOrgOutcome, +): Promise { + try { + operation = await beginCompensation(operation, leaseToken, reason, outcome); + } catch { + // A successor owns the operation, or PostgreSQL cannot prove durable + // compensation intent. In either case this worker must not delete. + return false; + } + let organizationId = operation.workos_organization_id; + if (!organizationId) { + try { + const provider = validateProviderOrganization( + await workos.organizations.getOrganizationByExternalId(operation.provider_external_id), + ); + organizationId = provider.id; + operation = await persistOperation(operation.id, leaseToken, { + status: 'compensating', + workosOrganizationId: provider.id, + workosOrganizationName: provider.name, + errorCode: reason, + }); + } catch (error) { + if (isProviderNotFound(error)) { + return markOperation(operation.id, leaseToken, 'failed', reason).catch(() => false); + } + await persistOperation(operation.id, leaseToken, { + status: 'compensating', errorCode: `${reason}:lookup_unavailable`, releaseLease: true, + }).catch(() => {}); + return false; + } + } + const local = await getPool().query( + 'SELECT 1 FROM organizations WHERE workos_organization_id = $1', + [organizationId], + ).catch(() => null); + if (local === null || local.rowCount) { + await markOperation(operation.id, leaseToken, 'manual_reconciliation', `${reason}:local_state_uncertain`).catch(() => {}); + return false; + } + try { + await renewOperationLease(operation.id, leaseToken); + } catch { + return false; + } + try { + await workos.organizations.deleteOrganization(organizationId); + return markOperation(operation.id, leaseToken, 'failed', reason).catch(() => false); + } catch (error) { + if (isProviderNotFound(error)) { + return markOperation(operation.id, leaseToken, 'failed', reason).catch(() => false); + } + try { + await workos.organizations.getOrganizationByExternalId(operation.provider_external_id); + } catch (checkError) { + if (isProviderNotFound(checkError)) { + return markOperation(operation.id, leaseToken, 'failed', reason).catch(() => false); + } + } + await persistOperation(operation.id, leaseToken, { + status: 'compensating', errorCode: `${reason}:delete_unconfirmed`, releaseLease: true, + }).catch(() => {}); + return false; + } +} + +async function executeOperation( + workos: WorkOS, + operation: OnboardingOperation, + leaseToken: string, +): Promise { + if (operation.status === 'compensating') { + const outcome = terminalFailureOutcome(operation); + const cleaned = await compensateProviderOrganization( + workos, operation, leaseToken, operation.last_error_code || outcome.kind, outcome, + ); + return cleaned ? outcome + : { kind: 'onboarding_reconciliation_required', operationId: operation.id }; + } + try { + operation = await ensureProviderOrganization(workos, operation, leaseToken); + operation = await ensureProviderMembership(workos, operation, leaseToken); + return await finalizeLocalOperation(workos, operation, leaseToken); + } catch (error) { + if (error instanceof FinalizationBusinessError) { + const cleaned = await compensateProviderOrganization( + workos, operation, leaseToken, error.outcome.kind, error.outcome, + ); + return cleaned ? error.outcome + : { kind: 'onboarding_reconciliation_required', operationId: operation.id }; + } + if (error instanceof AuthorityChangedError) { + const terminal: CreateOrgOutcome = { kind: 'credential_changed' }; + const cleaned = await compensateProviderOrganization( + workos, operation, leaseToken, terminal.kind, terminal, + ); + return cleaned ? { kind: 'credential_changed' } + : { kind: 'onboarding_reconciliation_required', operationId: operation.id }; + } + if (error instanceof ProviderDivergenceError) { + await markOperation(operation.id, leaseToken, 'manual_reconciliation', 'provider_state_diverged').catch(() => {}); + return { kind: 'onboarding_reconciliation_required', operationId: operation.id }; + } + if (error instanceof ProviderUnavailableError || isTransientDatabaseFailure(error)) { + await persistOperation(operation.id, leaseToken, { + errorCode: 'retryable_provider_or_database_failure', releaseLease: true, + }).catch(() => {}); + return { kind: 'onboarding_retryable', operationId: operation.id }; + } + logger.error({ error, operationId: operation.id }, 'Organization onboarding failed'); + const terminal: CreateOrgOutcome = { kind: 'authorization_unavailable' }; + const cleaned = await compensateProviderOrganization( + workos, operation, leaseToken, 'non_retryable_provider_failure', terminal, + ); + return cleaned ? { kind: 'authorization_unavailable' } + : { kind: 'onboarding_reconciliation_required', operationId: operation.id }; + } +} export async function performCreateOrganization( - _input: CreateOrgRequest, - _deps: { workos: WorkOS | null; orgDb: OrganizationDatabase }, + input: CreateOrgRequest, + deps: { workos: WorkOS | null }, ): Promise { - return { kind: 'onboarding_disabled' }; + const validated = validateInput(input); + if ('kind' in validated) return validated; + if (input.user.impersonator) return { kind: 'impersonation_not_allowed' }; + const actorId = input.user.authWorkosUserId ?? input.user.id; + const credentialError = await verifyPresentedCredential(input, actorId); + if (credentialError) return credentialError; + const snapshot = input.user.authorizationSnapshot; + if (!snapshot) return { kind: 'authorization_unavailable' }; + if (!deps.workos) return { kind: 'authorization_unavailable' }; + + let provider: ProviderIdentity; + try { + provider = await fetchProviderIdentity(deps.workos, actorId); + } catch (error) { + return error instanceof AuthorityChangedError || error instanceof ProviderDivergenceError + ? { kind: 'credential_changed' } + : { kind: 'authorization_unavailable' }; + } + + const prepared = await prepareOperation(input, validated, provider, actorId, snapshot); + if (prepared.kind === 'outcome') return prepared.outcome; + if (prepared.kind === 'cleanup') { + const cleaned = await compensateProviderOrganization( + deps.workos, prepared.operation, prepared.leaseToken, 'superseded_or_changed_request', + { kind: 'credential_changed' }, + ); + return cleaned ? { kind: 'credential_changed' } + : { kind: 'onboarding_reconciliation_required', operationId: prepared.operation.id }; + } + return executeOperation(deps.workos, prepared.operation, prepared.leaseToken); +} + +export type OrganizationOnboardingReconciliationResult = { + attempted: number; + completed: number; + compensated: number; + retryable: number; + manual: number; +}; + +export type ManualOrganizationOnboardingOperation = { + id: string; + authenticatedWorkosUserId: string; + workosOrganizationId: string | null; + lastErrorCode: string | null; + createdAt: Date; + updatedAt: Date; +}; + +/** + * Return durable manual-reconciliation records for the operator escalation + * worker. These rows are deliberately not executed automatically: their + * provider/local authority state is ambiguous and requires human review. + */ +export async function listManualOrganizationOnboardingOperations( + limit = 10, +): Promise { + const boundedLimit = Math.max(0, Math.min(limit, 25)); + if (boundedLimit === 0) return []; + const result = await getPool().query<{ + id: string; + authenticated_workos_user_id: string; + workos_organization_id: string | null; + last_error_code: string | null; + created_at: Date; + updated_at: Date; + }>( + `SELECT operation.id, operation.authenticated_workos_user_id, + operation.workos_organization_id, operation.last_error_code, + operation.created_at, operation.updated_at + FROM organization_onboarding_operations operation + WHERE operation.status = 'manual_reconciliation' + AND NOT EXISTS ( + SELECT 1 + FROM addie_escalations escalation + WHERE escalation.dedup_key = + 'organization-onboarding-reconciliation:' || operation.id::text + AND escalation.status IN ('open', 'acknowledged', 'in_progress') + ) + ORDER BY operation.updated_at, operation.created_at + LIMIT $1`, + [boundedLimit], + ); + return result.rows.map(row => ({ + id: row.id, + authenticatedWorkosUserId: row.authenticated_workos_user_id, + workosOrganizationId: row.workos_organization_id, + lastErrorCode: row.last_error_code, + createdAt: row.created_at, + updatedAt: row.updated_at, + })); +} + +async function claimReconciliationOperation(excludedIds: string[]): Promise<{ + operation: OnboardingOperation; + leaseToken: string; +} | null> { + const leaseToken = randomUUID(); + const result = await getPool().query( + `UPDATE organization_onboarding_operations operation + SET lease_token = $2, + lease_expires_at = NOW() + ($3::text || ' seconds')::interval, + updated_at = NOW() + WHERE operation.id = ( + SELECT candidate.id + FROM organization_onboarding_operations candidate + WHERE candidate.status IN ( + 'pending_provider_organization', + 'pending_provider_membership', + 'pending_local_commit', + 'compensating' + ) + AND (candidate.lease_token IS NULL OR candidate.lease_expires_at <= NOW()) + AND NOT (candidate.id = ANY($1::uuid[])) + ORDER BY candidate.updated_at, candidate.created_at + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING operation.*`, + [excludedIds, leaseToken, String(LEASE_SECONDS)], + ); + return result.rows[0] ? { operation: result.rows[0], leaseToken } : null; +} + +/** + * Resume provider/local operations whose request worker stopped or returned a + * retryable response. Each operation is fenced by the same durable lease and + * revalidates the exact current credential before any authority commit. + */ +export async function reconcilePendingOrganizationOnboarding( + workos: WorkOS, + limit = 2, +): Promise { + const result: OrganizationOnboardingReconciliationResult = { + attempted: 0, completed: 0, compensated: 0, retryable: 0, manual: 0, + }; + const attemptedIds: string[] = []; + for (let index = 0; index < Math.max(0, Math.min(limit, 10)); index++) { + const claimed = await claimReconciliationOperation(attemptedIds); + if (!claimed) break; + attemptedIds.push(claimed.operation.id); + result.attempted++; + const outcome = await executeOperation(workos, claimed.operation, claimed.leaseToken); + if (outcome.kind === 'created') result.completed++; + else if (outcome.kind === 'onboarding_retryable') result.retryable++; + else if (outcome.kind === 'onboarding_reconciliation_required') result.manual++; + else result.compensated++; + } + return result; } diff --git a/server/src/services/organization-membership-mutation.ts b/server/src/services/organization-membership-mutation.ts index 4ff06e28576..7afad2e0a11 100644 --- a/server/src/services/organization-membership-mutation.ts +++ b/server/src/services/organization-membership-mutation.ts @@ -9,6 +9,7 @@ import { isInvalidWorkOSJWTError, verifyWorkOSJWT } from '../auth/workos-jwt.js' import { DEV_USERS, isDevModeEnabled, invalidateSessionsForUsers } from '../middleware/auth.js'; import { getSeatLimits, resolveMembershipTier, type SeatType, type MembershipTierRow } from '../db/organization-db.js'; import { createLogger } from '../logger.js'; +import { lockExactCredentialAuthorizationRows } from './exact-credential-authorization-lock.js'; const logger = createLogger('organization-membership-mutation'); export type Role = 'owner' | 'admin' | 'member'; @@ -255,13 +256,8 @@ export class OrganizationMembershipMutation { // Existing identity/webhook writers take these rows in differing orders. // Never wait while holding a competing row: fail closed on contention, // rather than forming a cycle after an irreversible provider action. - await this.db.query('SELECT workos_user_id FROM users WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', [ids]); - await this.db.query('SELECT workos_user_id FROM authorization_epochs WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', [ids]); - await this.db.query('SELECT workos_user_id FROM identity_workos_users WHERE workos_user_id = ANY($1) ORDER BY workos_user_id FOR UPDATE NOWAIT', [ids]); + await lockExactCredentialAuthorizationRows(this.db, ids); await this.db.query('SELECT workos_user_id FROM organization_memberships WHERE workos_organization_id = $1 ORDER BY workos_user_id FOR UPDATE NOWAIT', [this.orgId]); - // Ban insertion has no existing row to lock. This short shared table lock - // also covers deletes/expiry updates until the local transaction commits. - await this.db.query('LOCK TABLE bans IN SHARE MODE NOWAIT'); await this.checkProviderAuthority(); await this.checkTargets(); this.localPhase = true; @@ -498,10 +494,7 @@ export async function cancelJoinRequestForExactCredential(req: Request, requestI mutationDenied('Authentication state changed; retry the request'); } - await db.query('SELECT workos_user_id FROM users WHERE workos_user_id = $1 FOR UPDATE NOWAIT', [actorId]); - await db.query('SELECT workos_user_id FROM authorization_epochs WHERE workos_user_id = $1 FOR UPDATE NOWAIT', [actorId]); - await db.query('SELECT workos_user_id FROM identity_workos_users WHERE workos_user_id = $1 FOR UPDATE NOWAIT', [actorId]); - await db.query('LOCK TABLE bans IN SHARE MODE NOWAIT'); + await lockExactCredentialAuthorizationRows(db, [actorId]); const current = await readCancellationSnapshot(db, actorId, orgId); if (current.banned) mutationDenied('Account suspended'); if (JSON.stringify(current) !== JSON.stringify(snapshot)) mutationDenied('Authorization changed; retry the request'); diff --git a/server/tests/integration/claim-prospect-org.test.ts b/server/tests/integration/claim-prospect-org.test.ts index 1c5dac751b1..211004416fd 100644 --- a/server/tests/integration/claim-prospect-org.test.ts +++ b/server/tests/integration/claim-prospect-org.test.ts @@ -171,8 +171,8 @@ describe('POST /api/organizations/:orgId/claim', () => { userOverride = { emailVerified: verified as boolean, email: `claimer@${domain}` }; await seedProspect(pool, { orgId: TEST_ORG_PROSPECT, domain: TEST_DOMAIN }); const res = await request(app).post(`/api/organizations/${TEST_ORG_PROSPECT}/claim`).send({}); - expect(res.status).toBe(403); - expect(res.body.error).toBe('organization_onboarding_disabled'); + expect(res.status).toBe(409); + expect(res.body.error).toBe('organization_adoption_unavailable'); expect(mockCreateOrganizationMembership).not.toHaveBeenCalled(); expect((await pool.query('SELECT * FROM organization_memberships WHERE workos_organization_id = $1', [TEST_ORG_PROSPECT])).rowCount).toBe(0); expect((await pool.query('SELECT * FROM registry_audit_log WHERE workos_organization_id = $1', [TEST_ORG_PROSPECT])).rowCount).toBe(0); diff --git a/server/tests/integration/implicit-onboarding-containment.test.ts b/server/tests/integration/implicit-onboarding-containment.test.ts index 5a4a687b449..08673d9a0da 100644 --- a/server/tests/integration/implicit-onboarding-containment.test.ts +++ b/server/tests/integration/implicit-onboarding-containment.test.ts @@ -7,7 +7,7 @@ const mocks = vi.hoisted(() => { return { sessions: new Map(), providerMemberships: new Map(), - sendInvitation: vi.fn(), getInvitation: vi.fn(), revokeInvitation: vi.fn(), + sendInvitation: vi.fn(), getInvitation: vi.fn(), acceptInvitation: vi.fn(), revokeInvitation: vi.fn(), listUsers: vi.fn(), list: vi.fn(), getUser: vi.fn(), create: vi.fn(), update: vi.fn(), createOrg: vi.fn(), invoice: vi.fn(), products: vi.fn(), coupon: vi.fn(), slackSync: vi.fn(), slackConfigured: false, @@ -21,7 +21,8 @@ vi.mock('@workos-inc/node', () => ({ WorkOS: class { return { authenticated: true, user, accessToken: `test-access:${user.id}` }; } }), listOrganizationMemberships: mocks.list, - sendInvitation: mocks.sendInvitation, getInvitation: mocks.getInvitation, revokeInvitation: mocks.revokeInvitation, + sendInvitation: mocks.sendInvitation, getInvitation: mocks.getInvitation, + acceptInvitation: mocks.acceptInvitation, revokeInvitation: mocks.revokeInvitation, listUsers: mocks.listUsers, getUser: mocks.getUser, createOrganizationMembership: mocks.create, @@ -78,14 +79,18 @@ const A = 'user_containment_a'; const B = 'user_containment_b'; const ORG = 'org_containment'; const TOKEN = 'private-containment-token'; -const routes: Array<[string, Record]> = [ - ['/api/organizations', { organization_name: 'Acme Containment', is_personal: false }], - [`/api/organizations/${ORG}/claim`, {}], - ['/api/join-requests', { organization_id: ORG }], - ['/api/me/agents', { url: 'https://agent.containment.test/mcp', type: 'sales' }], - [`/api/invite/${TOKEN}/accept`, { agreement_version: '999.0', billingAddress: { - line1: '1 Example St', city: 'Example', state: 'EX', postal_code: '10000', country: 'US', - } }], +const AGENT_BODY = { url: 'https://agent.containment.test/mcp', type: 'sales' }; +const TOKEN_ACCEPT_PATH = `/api/invite/${TOKEN}/accept`; +const TOKEN_ACCEPT_BODY = { agreement_version: '999.0', billingAddress: { + line1: '1 Example St', city: 'Example', state: 'EX', postal_code: '10000', country: 'US', +} }; +const routes: Array<[string, Record, number, string]> = [ + ['/api/organizations', { organization_name: 'Acme Containment', is_personal: false }, 400, 'agreement_acceptance_required'], + [`/api/organizations/${ORG}/claim`, {}, 409, 'organization_adoption_unavailable'], + ['/api/join-requests', { organization_id: ORG }, 403, 'organization_join_onboarding_unavailable'], + ['/api/invitations/inv_containment/accept', {}, 403, 'organization_invitation_acceptance_unavailable'], + ['/api/me/agents', AGENT_BODY, 400, 'explicit_organization_required'], + [TOKEN_ACCEPT_PATH, TOKEN_ACCEPT_BODY, 403, 'organization_onboarding_disabled'], ]; let pool: Pool; let app: HTTPServer['app']; @@ -120,7 +125,8 @@ async function link(primary: string, sibling: string) { invalidateSessionsForUsers([A, B]); } function expectNoProviders() { - for (const spy of [mocks.create, mocks.update, mocks.createOrg, mocks.sendInvitation, mocks.invoice, mocks.coupon, mocks.products]) { + for (const spy of [mocks.create, mocks.update, mocks.createOrg, mocks.sendInvitation, + mocks.acceptInvitation, mocks.invoice, mocks.coupon, mocks.products]) { expect(spy).not.toHaveBeenCalled(); } } @@ -326,26 +332,27 @@ describe('mounted implicit onboarding containment', () => { expect((await request(app).post(path).send(body)).status).toBe(401); expectNoProviders(); }); - it.each(routes)('linked B proof never grants canonical A via %s', async (path, body) => { + it.each(routes)('linked B proof never grants canonical A via %s', async (path, body, expectedStatus, expectedError) => { await link(A, B); const before = await snapshot(); const res = await request(app).post(path).set('Cookie', cookie()).send(body); expectNoProviders(); - expect(res.status).toBe(403); - expect(res.body.error).toBe('organization_onboarding_disabled'); + expect(res.status).toBe(expectedStatus); + expect(res.body.error).toBe(expectedError); expect(await snapshot()).toEqual(before); }); it.each([true, false, 'missing'] as const)('verification %s cannot enable suspended grants', async (verified) => { - for (const [path, body] of routes) { + for (const [path, body, expectedStatus, expectedError] of routes) { const res = await request(app).post(path).set('Cookie', cookie(B, verified)).send(body); - expect(res.status).toBe(403); + expect(res.status).toBe(expectedStatus); + expect(res.body.error).toBe(expectedError); } expectNoProviders(); }); it.each([' B@CONTAINMENT.TEST ', 'b@子.test', 'alias@containment.test', 'a@unrelated.test'])('token email %s does not bypass denial', async (email) => { const before = await snapshot(); - const [path, body] = routes[4]; - expect((await request(app).post(path).set('Cookie', cookie(B, true, email)).send(body)).status).toBe(403); + expect((await request(app).post(TOKEN_ACCEPT_PATH).set('Cookie', cookie(B, true, email)) + .send(TOKEN_ACCEPT_BODY)).status).toBe(403); expect(await snapshot()).toEqual(before); expectNoProviders(); }); @@ -354,8 +361,8 @@ describe('mounted implicit onboarding containment', () => { if (status === 'accepted') await pool.query('UPDATE membership_invites SET accepted_at = NOW(), accepted_by_user_id = $2 WHERE token = $1', [TOKEN, B]); if (status === 'expired') await pool.query("UPDATE membership_invites SET expires_at = NOW() - interval '1 day' WHERE token = $1", [TOKEN]); const before = await snapshot(); - const [path, body] = routes[4]; - const results = await Promise.all([A, B].map(id => request(app).post(path).set('Cookie', cookie(id)).send(body))); + const results = await Promise.all([A, B].map(id => request(app).post(TOKEN_ACCEPT_PATH) + .set('Cookie', cookie(id)).send(TOKEN_ACCEPT_BODY))); expect(results.map(r => r.status)).toEqual([403, 403]); expect(await snapshot()).toEqual(before); expectNoProviders(); @@ -363,9 +370,13 @@ describe('mounted implicit onboarding containment', () => { it('opposite linked direction, concurrent independent sessions and unlink preserve zero transferred authority', async () => { await link(B, A); const before = await snapshot(); - const requests = routes.flatMap(([path, body]) => [A, B].map(id => - request(app).post(path).set('Cookie', cookie(id)).send(body))); - expect((await Promise.all(requests)).every(r => r.status === 403)).toBe(true); + const requests = routes.flatMap(([path, body, expectedStatus, expectedError]) => [A, B].map(async id => { + const response = await request(app).post(path).set('Cookie', cookie(id)).send(body); + expect(response.status).toBe(expectedStatus); + expect(response.body.error).toBe(expectedError); + return response; + })); + await Promise.all(requests); await pool.query('DELETE FROM identity_workos_users WHERE workos_user_id = $1', [A]); invalidateSessionsForUsers([A, B]); expect(await snapshot()).toEqual(before); @@ -375,7 +386,11 @@ describe('mounted implicit onboarding containment', () => { mocks.list.mockRejectedValue(new Error('required source unavailable')); mocks.getUser.mockRejectedValue(new Error('required source unavailable')); const before = await snapshot(); - for (const [path, body] of routes) expect((await request(app).post(path).set('Cookie', cookie()).send(body)).status).toBe(403); + for (const [path, body, expectedStatus, expectedError] of routes) { + const response = await request(app).post(path).set('Cookie', cookie()).send(body); + expect(response.status).toBe(expectedStatus); + expect(response.body.error).toBe(expectedError); + } expect(await snapshot()).toEqual(before); expectNoProviders(); }); @@ -383,18 +398,19 @@ describe('mounted implicit onboarding containment', () => { for (let i = 0; i < count; i++) await pool.query(`INSERT INTO organization_memberships (workos_user_id, workos_organization_id, email, role) VALUES ($1, $2, 'b@containment.test', 'owner')`, [B, `${ORG}_${i}`]); const before = await snapshot(); - const res = await request(app).post('/api/me/agents').set('Cookie', cookie()).send(routes[3][1]); - expect(res.status).toBe(403); + const res = await request(app).post('/api/me/agents').set('Cookie', cookie()).send(AGENT_BODY); + expect(res.status).toBe(400); + expect(res.body.error).toBe('explicit_organization_required'); expect(await snapshot()).toEqual(before); expectNoProviders(); }); it('explicit agent organization denies canonical sibling membership and reports provider outage as 503', async () => { await link(A, B); mocks.list.mockResolvedValue({ data: [{ userId: A, organizationId: ORG, status: 'active', role: { slug: 'owner' } }] }); - expect((await request(app).post('/api/me/agents').query({ org: ORG }).set('Cookie', cookie()).send(routes[3][1])).status).toBe(403); + expect((await request(app).post('/api/me/agents').query({ org: ORG }).set('Cookie', cookie()).send(AGENT_BODY)).status).toBe(403); expect(mocks.list).toHaveBeenCalledWith({ userId: B, organizationId: ORG, statuses: ['active'] }); mocks.list.mockRejectedValue(new Error('outage')); - expect((await request(app).post('/api/me/agents').query({ org: ORG }).set('Cookie', cookie()).send(routes[3][1])).status).toBe(503); + expect((await request(app).post('/api/me/agents').query({ org: ORG }).set('Cookie', cookie()).send(AGENT_BODY)).status).toBe(503); expectNoProviders(); }); it.each([ diff --git a/server/tests/integration/join-request-approval.test.ts b/server/tests/integration/join-request-approval.test.ts index e773275f4e3..8ec5a15b454 100644 --- a/server/tests/integration/join-request-approval.test.ts +++ b/server/tests/integration/join-request-approval.test.ts @@ -174,7 +174,7 @@ describe('Join Request Approval', () => { .send({ organization_id: TEST_ORG_ID }) .expect(403); - expect(response.body.error).toBe('organization_onboarding_disabled'); + expect(response.body.error).toBe('organization_join_onboarding_unavailable'); expect(mockCreateOrganizationMembership).not.toHaveBeenCalled(); }); @@ -195,7 +195,7 @@ describe('Join Request Approval', () => { .send({ organization_id: TEST_ORG_ID }) .expect(403); - expect(response.body.error).toBe('organization_onboarding_disabled'); + expect(response.body.error).toBe('organization_join_onboarding_unavailable'); const result = await pool.query( `SELECT id FROM organization_join_requests diff --git a/server/tests/integration/member-agents-auto-bootstrap.test.ts b/server/tests/integration/member-agents-auto-bootstrap.test.ts index 73a43487e47..e2ea4dc8318 100644 --- a/server/tests/integration/member-agents-auto-bootstrap.test.ts +++ b/server/tests/integration/member-agents-auto-bootstrap.test.ts @@ -295,8 +295,8 @@ describe('POST /api/me/agents (auto-bootstrap)', () => { const before = (await pool.query('SELECT * FROM organization_memberships WHERE workos_user_id = $1', [USER_ID])).rows; const res = await request(app).post('/api/me/agents') .send({ url: 'https://agent.boot-corp.test/mcp', type: 'sales', visibility: 'private' }); - expect(res.status).toBe(403); - expect(res.body.error).toBe('organization_onboarding_disabled'); + expect(res.status).toBe(400); + expect(res.body.error).toBe('explicit_organization_required'); expect((await pool.query('SELECT * FROM organization_memberships WHERE workos_user_id = $1', [USER_ID])).rows).toEqual(before); }); }); diff --git a/server/tests/integration/organization-onboarding-restoration.test.ts b/server/tests/integration/organization-onboarding-restoration.test.ts new file mode 100644 index 00000000000..210250b434f --- /dev/null +++ b/server/tests/integration/organization-onboarding-restoration.test.ts @@ -0,0 +1,1080 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import request from 'supertest'; + +const mocks = vi.hoisted(() => { + process.env.WORKOS_API_KEY = 'sk_test_onboarding'; + process.env.WORKOS_CLIENT_ID = 'client_test_onboarding'; + process.env.WORKOS_COOKIE_PASSWORD = 'x'.repeat(32); + process.env.WORKOS_WEBHOOK_SECRET = 'onboarding-restoration-test-secret'; + return { + sessions: new Map(), + providerUsers: new Map(), + providerOrganizations: new Map(), + providerMemberships: new Map(), + getUser: vi.fn(), + getOrganizationByExternalId: vi.fn(), + createOrganization: vi.fn(), + deleteOrganization: vi.fn(), + listOrganizationMemberships: vi.fn(), + createOrganizationMembership: vi.fn(), + createApiKeyValidation: vi.fn(), + }; +}); + +function notFound(): Error { + return Object.assign(new Error('not found'), { status: 404 }); +} + +vi.mock('@workos-inc/node', () => ({ + WorkOS: class { + userManagement = { + loadSealedSession: ({ sessionData }: any) => ({ + authenticate: async () => { + const stored = mocks.sessions.get(sessionData); + const user = stored.user || stored; + return { + authenticated: true, + user, + accessToken: `test-access:${user.id}`, + impersonator: stored.impersonator, + }; + }, + }), + getUser: mocks.getUser, + listOrganizationMemberships: mocks.listOrganizationMemberships, + createOrganizationMembership: mocks.createOrganizationMembership, + }; + organizations = { + getOrganizationByExternalId: mocks.getOrganizationByExternalId, + createOrganization: mocks.createOrganization, + deleteOrganization: mocks.deleteOrganization, + }; + apiKeys = { createValidation: mocks.createApiKeyValidation }; + webhooks = { constructEvent: vi.fn().mockResolvedValue({}) }; + }, +})); + +vi.mock('../../src/auth/workos-client.js', async (original) => { + const actual = await original(); + const { WorkOS } = await import('@workos-inc/node'); + const instance = new WorkOS(); + return { + ...actual, + getWorkos: () => instance, + getAuthorizationEnforcementWorkos: () => instance, + }; +}); + +vi.mock('../../src/auth/workos-jwt.js', async (original) => ({ + ...await original(), + verifyWorkOSJWT: async (value: string) => ({ + sub: value === 'm2m.m2m.m2m' ? 'client_onboarding_test' : value.replace(/^test-access:/, ''), + isM2M: value === 'm2m.m2m.m2m', + expiresAt: Math.floor(Date.now() / 1000) + 300, + }), +})); + +vi.mock('../../src/middleware/csrf.js', () => ({ + csrfProtection: (_req: any, _res: any, next: any) => next(), +})); + +vi.mock('../../src/middleware/rate-limit.js', async (original) => { + const actual = await original(); + const pass = (_req: any, _res: any, next: any) => next(); + return { ...actual, orgCreationRateLimiter: pass }; +}); + +vi.mock('../../src/addie/error-notifier.js', () => ({ notifySystemError: vi.fn() })); + +import { HTTPServer } from '../../src/http.js'; +import { closeDatabase, initializeDatabase } from '../../src/db/client.js'; +import { runMigrations } from '../../src/db/migrate.js'; +import { invalidateSessionsForUsers } from '../../src/middleware/auth.js'; +import { reconcilePendingOrganizationOnboarding } from '../../src/services/organization-bootstrap.js'; +import { getAuthorizationEnforcementWorkos } from '../../src/auth/workos-client.js'; +import { runOrganizationOnboardingReconciliationJob } from '../../src/addie/jobs/organization-onboarding-reconciliation.js'; +import type { Pool } from 'pg'; + +const A = 'user_onboarding_primary'; +const B = 'user_onboarding_exact'; +let pool: Pool; +let app: HTTPServer['app']; +let sessionNumber = 0; +let providerOrgNumber = 0; + +function cookie(id = B, impersonator?: { email: string; reason: string | null }): string { + const key = `onboarding-session-${++sessionNumber}`; + const provider = mocks.providerUsers.get(id); + const user = { + id, + email: provider.email, + emailVerified: provider.emailVerified, + firstName: provider.firstName, + lastName: provider.lastName, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + mocks.sessions.set(key, impersonator ? { user, impersonator } : user); + return `wos-session=${key}`; +} + +function createBody(overrides: Record = {}) { + return { + organization_name: 'Nova Onboarding', + company_type: 'brand', + revenue_tier: 'under_1m', + marketing_opt_in: true, + agreements_accepted: true, + terms_version: '1.0', + privacy_version: '1.0', + ...overrides, + }; +} + +function exactMembership(userId: string, organizationId: string, id = `om_${organizationId}`) { + return { + id, + userId, + organizationId, + status: 'active', + role: { slug: 'owner' }, + }; +} + +async function link(primary: string, sibling: string) { + await pool.query('UPDATE identity_workos_users SET is_primary = false WHERE workos_user_id = $1', [sibling]); + await pool.query( + `UPDATE identity_workos_users SET identity_id = + (SELECT identity_id FROM identity_workos_users WHERE workos_user_id = $1) + WHERE workos_user_id = $2`, + [primary, sibling], + ); + await pool.query('UPDATE identity_workos_users SET is_primary = true WHERE workos_user_id = $1', [primary]); + invalidateSessionsForUsers([primary, sibling]); +} + +async function localAuthorityRows() { + const [organizations, memberships, domains, audits, agreements, epochs, operations] = await Promise.all([ + pool.query("SELECT * FROM organizations WHERE workos_organization_id LIKE 'org_onboarding_%' ORDER BY workos_organization_id"), + pool.query('SELECT * FROM organization_memberships WHERE workos_user_id = ANY($1) ORDER BY workos_organization_id', [[A, B]]), + pool.query("SELECT * FROM organization_domains WHERE workos_organization_id LIKE 'org_onboarding_%' ORDER BY id"), + pool.query("SELECT * FROM registry_audit_log WHERE resource_id LIKE 'org_onboarding_%' ORDER BY created_at"), + pool.query('SELECT * FROM user_agreement_acceptances WHERE workos_user_id = ANY($1) ORDER BY agreement_type', [[A, B]]), + pool.query('SELECT * FROM authorization_epochs WHERE workos_user_id = ANY($1) ORDER BY workos_user_id', [[A, B]]), + pool.query('SELECT * FROM organization_onboarding_operations WHERE authenticated_workos_user_id = ANY($1) ORDER BY created_at', [[A, B]]), + ]); + return { + organizations: organizations.rows, + memberships: memberships.rows, + domains: domains.rows, + audits: audits.rows, + agreements: agreements.rows, + epochs: epochs.rows, + operations: operations.rows, + }; +} + +async function cleanupTestRows() { + await pool.query( + `DELETE FROM addie_escalations + WHERE dedup_key LIKE 'organization-onboarding-reconciliation:%' + AND workos_user_id = ANY($1)`, + [[A, B]], + ); + await pool.query("DELETE FROM registry_audit_log WHERE resource_id LIKE 'org_onboarding_%'"); + await pool.query('DELETE FROM organization_memberships WHERE workos_user_id = ANY($1)', [[A, B]]); + await pool.query("DELETE FROM organizations WHERE workos_organization_id = 'org_existing_prospect'"); + await pool.query("DELETE FROM organizations WHERE workos_organization_id LIKE 'org_onboarding_%'"); + await pool.query('DELETE FROM user_agreement_acceptances WHERE workos_user_id = ANY($1)', [[A, B]]); + await pool.query('DELETE FROM organization_onboarding_operations WHERE authenticated_workos_user_id = ANY($1)', [[A, B]]); + await pool.query('DELETE FROM user_email_preferences WHERE workos_user_id = ANY($1)', [[A, B]]); + await pool.query('DELETE FROM users WHERE workos_user_id = ANY($1)', [[A, B]]); +} + +beforeAll(async () => { + const connectionString = process.env.DATABASE_URL; + if (!connectionString || !/^postgres(?:ql)?:\/\/[^/]+@(127\.0\.0\.1|localhost):\d+\/adcp_test$/.test(connectionString)) { + throw new Error('This suite requires an explicit loopback adcp_test database'); + } + pool = initializeDatabase({ connectionString }); + await runMigrations(); + app = new HTTPServer({ backgroundServices: 'refresh-only' }).app; +}, 60_000); + +beforeEach(async () => { + vi.clearAllMocks(); + providerOrgNumber = 0; + invalidateSessionsForUsers([A, B]); + await cleanupTestRows(); + await pool.query( + `INSERT INTO users (workos_user_id, email, email_verified, first_name, last_name) + VALUES ($1, 'primary@unrelated.test', true, 'Primary', 'Person'), + ($2, 'owner@nova-onboarding.test', true, 'Exact', 'Owner')`, + [A, B], + ); + + mocks.providerOrganizations.clear(); + mocks.providerMemberships.clear(); + mocks.providerUsers.clear(); + mocks.providerUsers.set(A, { + id: A, email: 'primary@unrelated.test', emailVerified: true, + firstName: 'Primary', lastName: 'Person', + }); + mocks.providerUsers.set(B, { + id: B, email: 'owner@nova-onboarding.test', emailVerified: true, + firstName: 'Exact', lastName: 'Owner', + }); + mocks.getUser.mockImplementation(async (id: string) => { + const user = mocks.providerUsers.get(id); + if (!user) throw notFound(); + return { ...user }; + }); + mocks.getOrganizationByExternalId.mockImplementation(async (externalId: string) => { + const org = mocks.providerOrganizations.get(externalId); + if (!org) throw notFound(); + return { ...org }; + }); + mocks.createOrganization.mockImplementation(async (payload: any) => { + const org = { + id: `org_onboarding_${++providerOrgNumber}`, + name: payload.name, + externalId: payload.externalId, + }; + mocks.providerOrganizations.set(payload.externalId, org); + return { ...org }; + }); + mocks.deleteOrganization.mockImplementation(async (organizationId: string) => { + for (const [externalId, org] of mocks.providerOrganizations) { + if (org.id === organizationId) mocks.providerOrganizations.delete(externalId); + } + for (const [id, membership] of mocks.providerMemberships) { + if (membership.organizationId === organizationId) mocks.providerMemberships.delete(id); + } + }); + mocks.listOrganizationMemberships.mockImplementation(async ({ userId, organizationId, statuses }: any) => ({ + data: [...mocks.providerMemberships.values()].filter(row => + (!userId || row.userId === userId) + && (!organizationId || row.organizationId === organizationId) + && (!statuses || statuses.includes(row.status))), + })); + mocks.createOrganizationMembership.mockImplementation(async ({ userId, organizationId }: any) => { + const membership = exactMembership(userId, organizationId); + mocks.providerMemberships.set(membership.id, membership); + return { ...membership }; + }); + mocks.createApiKeyValidation.mockResolvedValue({ + apiKey: { + id: 'key_onboarding', + owner: { type: 'organization', id: 'org_api_key_onboarding' }, + name: 'Onboarding automation', + permissions: ['admin:*'], + }, + }); +}); + +afterAll(async () => { + await cleanupTestRows(); + await closeDatabase(); +}); + +describe('exact-credential organization onboarding restoration', () => { + it('creates a corporate organization with exact owner, verified domain, consent, marketing, audit, and epoch atomically', async () => { + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'corporate-success-0001') + .send(createBody({ + membership_tier: 'company_leader', + corporate_domain: 'attacker-controlled.example', + })); + + expect(response.status, JSON.stringify(response.body)).toBe(200); + expect(response.body).toMatchObject({ + success: true, + organization: { id: 'org_onboarding_1', name: 'Nova Onboarding' }, + }); + expect(response.body.operation_id).toEqual(expect.any(String)); + expect(mocks.createOrganization).toHaveBeenCalledWith( + expect.objectContaining({ + name: 'Nova Onboarding', + externalId: expect.stringMatching(/^adcp_onboarding_/), + metadata: { onboarding_operation_id: response.body.operation_id }, + }), + { idempotencyKey: expect.any(String) }, + ); + expect(mocks.createOrganizationMembership).toHaveBeenCalledWith({ + userId: B, + organizationId: 'org_onboarding_1', + roleSlug: 'owner', + }); + + const state = await localAuthorityRows(); + expect(state.organizations).toHaveLength(1); + expect(state.organizations[0]).toMatchObject({ + workos_organization_id: 'org_onboarding_1', + is_personal: false, + company_type: 'brand', + revenue_tier: 'under_1m', + membership_tier: null, + }); + expect(state.memberships).toHaveLength(1); + expect(state.memberships[0]).toMatchObject({ + workos_user_id: B, + workos_organization_id: 'org_onboarding_1', + role: 'owner', + workos_membership_id: 'om_org_onboarding_1', + }); + expect(state.domains).toEqual([ + expect.objectContaining({ + domain: 'nova-onboarding.test', verified: true, is_primary: true, + source: 'email_verification', + }), + ]); + expect(state.audits).toEqual([ + expect.objectContaining({ workos_user_id: B, action: 'organization_created' }), + ]); + expect(state.audits[0].details).toMatchObject({ + authenticated_workos_user_id: B, + canonical_workos_user_id: B, + authority: 'exact_credential_first_owner', + provider_identity_verified: true, + }); + expect(state.agreements.map(row => row.agreement_type).sort()).toEqual([ + 'privacy_policy', 'terms_of_service', + ]); + expect(state.epochs).toEqual([expect.objectContaining({ workos_user_id: B, epoch: '1' })]); + expect(state.operations).toEqual([ + expect.objectContaining({ status: 'completed', authenticated_workos_user_id: B }), + ]); + expect((await pool.query( + 'SELECT marketing_opt_in FROM user_email_preferences WHERE workos_user_id = $1', [B], + )).rows).toEqual([{ marketing_opt_in: true }]); + }); + + it('creates one personal workspace without assigning a corporate domain', async () => { + const body = createBody({ + organization_name: "Exact Owner's Workspace", + is_personal: true, + company_type: undefined, + revenue_tier: undefined, + marketing_opt_in: false, + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'personal-success-0001') + .send(body); + expect(response.status, JSON.stringify(response.body)).toBe(200); + const state = await localAuthorityRows(); + expect(state.organizations[0]).toMatchObject({ is_personal: true, email_domain: null }); + expect(state.domains).toEqual([]); + + invalidateSessionsForUsers([B]); + const duplicate = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'personal-second-0002') + .send({ ...body, organization_name: 'Another Personal Workspace' }); + expect(duplicate.status).toBe(409); + expect(duplicate.body.error).toBe('Personal workspace exists'); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + }); + + it('binds a linked sibling onboarding owner to the exact credential, never the canonical user', async () => { + await link(A, B); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie(B)) + .set('Idempotency-Key', 'linked-exact-owner-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status, JSON.stringify(response.body)).toBe(200); + expect(mocks.createOrganizationMembership).toHaveBeenCalledWith({ + userId: B, + organizationId: 'org_onboarding_1', + roleSlug: 'owner', + }); + const state = await localAuthorityRows(); + expect(state.memberships.map(row => row.workos_user_id)).toEqual([B]); + expect(state.audits.map(row => row.workos_user_id)).toEqual([B]); + expect(state.audits[0].details).toMatchObject({ + authenticated_workos_user_id: B, + canonical_workos_user_id: A, + }); + expect(state.epochs).toEqual([expect.objectContaining({ workos_user_id: B, epoch: '1' })]); + }); + + it('enforces the organization limit across every credential linked to the identity', async () => { + await link(A, B); + for (let index = 0; index < 10; index++) { + const orgId = `org_onboarding_limit_${index}`; + await pool.query( + `INSERT INTO organizations (workos_organization_id, name, is_personal) + VALUES ($1, $2, false)`, + [orgId, `Limit Organization ${index}`], + ); + await pool.query( + `INSERT INTO organization_memberships + (workos_user_id, workos_organization_id, email, role) + VALUES ($1, $2, 'primary@unrelated.test', 'owner')`, + [A, orgId], + ); + } + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie(B)) + .set('Idempotency-Key', 'linked-org-limit-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status).toBe(400); + expect(response.body.error).toBe('Organization limit reached'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + expect(mocks.createOrganizationMembership).not.toHaveBeenCalled(); + }); + + it('keeps prospect adoption contained with a distinct actionable response while making no provider write', async () => { + await pool.query( + `INSERT INTO organizations + (workos_organization_id, name, is_personal, email_domain, prospect_status) + VALUES ('org_existing_prospect', 'Nova Prospect', false, 'nova-onboarding.test', 'prospect') + ON CONFLICT (workos_organization_id) DO UPDATE SET prospect_status = 'prospect'`, + ); + await pool.query( + `INSERT INTO organization_domains + (workos_organization_id, domain, verified, is_primary, source) + VALUES ('org_existing_prospect', 'nova-onboarding.test', false, true, 'admin_discovery') + ON CONFLICT (domain) DO UPDATE SET workos_organization_id = EXCLUDED.workos_organization_id`, + ); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'prospect-contained-0001') + .send(createBody()); + expect(response.status).toBe(409); + expect(response.body).toMatchObject({ + error: 'organization_adoption_unavailable', + existing_org_id: 'org_existing_prospect', + existing_org_name: 'Nova Prospect', + }); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + expect(mocks.createOrganizationMembership).not.toHaveBeenCalled(); + await pool.query("DELETE FROM organizations WHERE workos_organization_id = 'org_existing_prospect'"); + }); + + it('requires current provider-verified corporate email and does not infer authority from the local email string', async () => { + mocks.providerUsers.set(B, { + ...mocks.providerUsers.get(B), emailVerified: false, + }); + invalidateSessionsForUsers([B]); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'unverified-domain-0001') + .send(createBody()); + expect(response.status).toBe(403); + expect(response.body.error).toBe('verified_corporate_email_required'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + }); + + it('binds consent to the exact displayed legal versions before any provider write', async () => { + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'stale-legal-version-0001') + .send(createBody({ terms_version: '0.9' })); + expect(response.status).toBe(400); + expect(response.body.error).toBe('agreement_acceptance_required'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + }); + + it('rejects impersonated first-owner and legal-consent actions', async () => { + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie(B, { email: 'admin@example.test', reason: 'support' })) + .set('Idempotency-Key', 'impersonated-owner-0001') + .send(createBody()); + expect(response.status).toBe(403); + expect(response.body.error).toBe('impersonation_not_allowed'); + expect(mocks.getUser).not.toHaveBeenCalled(); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + }); + + it('rejects API-key and M2M credentials before provider or local authorization effects', async () => { + const apiKeyResponse = await request(app).post('/api/organizations') + .set('Authorization', 'Bearer sk_onboarding_automation') + .set('Idempotency-Key', 'api-key-owner-0001') + .send(createBody()); + expect(apiKeyResponse.status).toBe(401); + expect(apiKeyResponse.body.error).toBe('invalid_credential'); + + const m2mResponse = await request(app).post('/api/organizations') + .set('Authorization', 'Bearer m2m.m2m.m2m') + .set('Idempotency-Key', 'm2m-owner-0001') + .send(createBody()); + expect(m2mResponse.status).toBe(401); + expect(m2mResponse.body.error).toBe('Invalid bearer token'); + + expect(mocks.getUser).not.toHaveBeenCalled(); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + expect(mocks.createOrganizationMembership).not.toHaveBeenCalled(); + expect((await localAuthorityRows()).operations).toEqual([]); + }); + + it('rejects malformed provider domains and canonicalizes IDNs to one ASCII ownership key', async () => { + mocks.providerUsers.set(B, { + ...mocks.providerUsers.get(B), email: 'owner@localhost', emailVerified: true, + }); + await pool.query('UPDATE users SET email = $2 WHERE workos_user_id = $1', [B, 'owner@localhost']); + invalidateSessionsForUsers([B]); + const malformed = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'malformed-domain-0001') + .send(createBody()); + expect(malformed.status).toBe(400); + expect(malformed.body.error).toBe('Corporate email required'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + + mocks.providerUsers.set(B, { + ...mocks.providerUsers.get(B), email: 'owner@hey.com', emailVerified: true, + }); + await pool.query('UPDATE users SET email = $2 WHERE workos_user_id = $1', [B, 'owner@hey.com']); + invalidateSessionsForUsers([B]); + const sharedMailbox = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'shared-mailbox-domain-0001') + .send(createBody()); + expect(sharedMailbox.status).toBe(400); + expect(sharedMailbox.body.error).toBe('Corporate email required'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + + mocks.providerUsers.set(B, { + ...mocks.providerUsers.get(B), email: 'owner@b\u00fccher.example', emailVerified: true, + }); + await pool.query('UPDATE users SET email = $2 WHERE workos_user_id = $1', [B, 'owner@b\u00fccher.example']); + invalidateSessionsForUsers([B]); + const idn = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'idn-domain-0001') + .send(createBody()); + expect(idn.status, JSON.stringify(idn.body)).toBe(200); + expect((await localAuthorityRows()).domains).toEqual([ + expect.objectContaining({ domain: 'xn--bcher-kva.example', verified: true }), + ]); + }); + + it('compensates a provider organization when the exact identity binding is replaced mid-operation', async () => { + const binding = (await pool.query<{ identity_id: string; is_primary: boolean }>( + 'SELECT identity_id, is_primary FROM identity_workos_users WHERE workos_user_id = $1', [B], + )).rows[0]; + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + await pool.query('DELETE FROM identity_workos_users WHERE workos_user_id = $1', [B]); + await pool.query( + `INSERT INTO identity_workos_users (workos_user_id, identity_id, is_primary) + VALUES ($1, $2, $3)`, + [B, binding.identity_id, binding.is_primary], + ); + return org; + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'binding-swap-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status).toBe(409); + expect(response.body.error).toBe('onboarding_credential_changed'); + expect(mocks.deleteOrganization).toHaveBeenCalledWith('org_onboarding_1'); + const state = await localAuthorityRows(); + expect(state.organizations).toEqual([]); + expect(state.memberships).toEqual([]); + expect(state.audits).toEqual([]); + expect(state.agreements).toEqual([]); + expect(state.epochs).toEqual([]); + expect(state.operations).toEqual([expect.objectContaining({ + status: 'failed', terminal_outcome: { kind: 'credential_changed' }, + })]); + + invalidateSessionsForUsers([B]); + const replay = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'binding-swap-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(replay.status).toBe(409); + expect(replay.body.error).toBe('onboarding_credential_changed'); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.deleteOrganization).toHaveBeenCalledTimes(1); + }); + + it('compensates when the provider email/domain changes after organization creation', async () => { + let reads = 0; + mocks.getUser.mockImplementation(async (id: string) => { + const user = { ...mocks.providerUsers.get(id) }; + if (++reads >= 3) user.email = 'owner@changed-domain.test'; + return user; + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'email-change-0001') + .send(createBody()); + expect(response.status).toBe(409); + expect(response.body.error).toBe('onboarding_credential_changed'); + expect(mocks.deleteOrganization).toHaveBeenCalledWith('org_onboarding_1'); + const state = await localAuthorityRows(); + expect(state.organizations).toEqual([]); + expect(state.memberships).toEqual([]); + expect(state.domains).toEqual([]); + expect(state.audits).toEqual([]); + expect(state.epochs).toEqual([]); + }); + + it('compensates when provider emailVerified changes before the local commit', async () => { + let reads = 0; + mocks.getUser.mockImplementation(async (id: string) => { + const user = { ...mocks.providerUsers.get(id) }; + if (++reads >= 3) user.emailVerified = false; + return user; + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'verification-change-0001') + .send(createBody()); + expect(response.status).toBe(409); + expect(response.body.error).toBe('onboarding_credential_changed'); + expect(mocks.deleteOrganization).toHaveBeenCalledWith('org_onboarding_1'); + const state = await localAuthorityRows(); + expect(state.organizations).toEqual([]); + expect(state.memberships).toEqual([]); + expect(state.domains).toEqual([]); + expect(state.audits).toEqual([]); + expect(state.epochs).toEqual([]); + }); + + it('fences stale compensation after an expired lease is claimed by a successor', async () => { + let reads = 0; + mocks.getUser.mockImplementation(async (id: string) => { + const user = { ...mocks.providerUsers.get(id) }; + if (++reads === 3) { + await pool.query( + `UPDATE organization_onboarding_operations + SET lease_token = gen_random_uuid(), lease_expires_at = NOW() - INTERVAL '1 second' + WHERE authenticated_workos_user_id = $1`, + [B], + ); + user.email = 'owner@changed-domain.test'; + } + return user; + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'expired-lease-fence-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status).toBe(503); + expect(response.body.error).toBe('organization_onboarding_reconciliation_required'); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + expect(mocks.providerOrganizations.size).toBe(1); + expect((await localAuthorityRows()).organizations).toEqual([]); + }); + + it('fences duplicate and changed concurrent submissions, then replays the completed response', async () => { + let releaseCreate!: () => void; + const createBlocked = new Promise(resolve => { releaseCreate = resolve; }); + let createStarted!: () => void; + const started = new Promise(resolve => { createStarted = resolve; }); + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + createStarted(); + await createBlocked; + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + return org; + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + const first = request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'concurrent-duplicate-0001') + .send(body).then(response => response); + await started; + const changed = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'concurrent-changed-0001') + .send({ ...body, organization_name: 'Changed While In Flight' }); + expect(changed.status).toBe(409); + expect(changed.body.error).toBe('organization_onboarding_in_progress'); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + + const duplicate = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'concurrent-duplicate-0001') + .send(body); + expect(duplicate.status).toBe(409); + expect(duplicate.body.error).toBe('organization_onboarding_in_progress'); + releaseCreate(); + const completed = await first; + expect(completed.status, JSON.stringify(completed.body)).toBe(200); + + invalidateSessionsForUsers([B]); + const replay = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'concurrent-duplicate-0001') + .send(body); + expect(replay.status).toBe(200); + expect(replay.body.organization).toEqual(completed.body.organization); + expect(replay.body.operation_id).toBe(completed.body.operation_id); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.createOrganizationMembership).toHaveBeenCalledTimes(1); + }); + + it('serializes concurrent personal-workspace creation across linked credentials', async () => { + await link(A, B); + let releaseCreate!: () => void; + const createBlocked = new Promise(resolve => { releaseCreate = resolve; }); + let createStarted!: () => void; + const started = new Promise(resolve => { createStarted = resolve; }); + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + createStarted(); + await createBlocked; + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + return org; + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + const first = request(app).post('/api/organizations') + .set('Cookie', cookie(B)) + .set('Idempotency-Key', 'linked-concurrent-b-0001') + .send(body).then(response => response); + await started; + const sibling = await request(app).post('/api/organizations') + .set('Cookie', cookie(A)) + .set('Idempotency-Key', 'linked-concurrent-a-0001') + .send({ ...body, organization_name: "Primary's Workspace" }); + expect(sibling.status).toBe(409); + expect(sibling.body.error).toBe('organization_onboarding_in_progress'); + releaseCreate(); + expect((await first).status).toBe(200); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.createOrganizationMembership).toHaveBeenCalledTimes(1); + expect((await localAuthorityRows()).memberships.map(row => row.workos_user_id)).toEqual([B]); + }); + + it('fails closed when a linked sibling credential is inside a concurrent membership mutation', async () => { + await link(A, B); + const blocker = await pool.connect(); + try { + await blocker.query('BEGIN'); + await blocker.query('SELECT workos_user_id FROM users WHERE workos_user_id = $1 FOR UPDATE', [A]); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie(B)) + .set('Idempotency-Key', 'sibling-membership-race-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status).toBe(503); + expect(response.body.error).toBe('authorization_unavailable'); + expect(mocks.createOrganization).not.toHaveBeenCalled(); + } finally { + await blocker.query('ROLLBACK'); + blocker.release(); + } + }); + + it('recovers an ambiguous provider organization create on retry by stable external ID', async () => { + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + throw Object.assign(new Error('timeout after write'), { code: 'ETIMEDOUT' }); + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + const first = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'org-timeout-retry-0001') + .send(body); + expect(first.status).toBe(503); + expect(first.body.error).toBe('organization_onboarding_retryable'); + expect((await localAuthorityRows()).organizations).toEqual([]); + + const retry = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'org-timeout-retry-0001') + .send(body); + expect(retry.status, JSON.stringify(retry.body)).toBe(200); + expect(retry.body.organization.id).toBe('org_onboarding_1'); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.createOrganizationMembership).toHaveBeenCalledTimes(1); + }); + + it('recovers when provider state was written and the database acknowledgement was lost', async () => { + const originalQuery = pool.query.bind(pool); + let injected = false; + const querySpy = vi.spyOn(pool as any, 'query').mockImplementation(async (...args: any[]) => { + const result = await (originalQuery as any)(...args); + const sql = typeof args[0] === 'string' ? args[0] : args[0]?.text; + const values = Array.isArray(args[1]) ? args[1] : args[0]?.values; + if (!injected && typeof sql === 'string' + && sql.includes('workos_organization_id = COALESCE') + && values?.[3] === 'org_onboarding_1') { + injected = true; + throw new Error('lost PostgreSQL acknowledgement after commit'); + } + return result; + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + let first; + try { + first = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'db-ack-loss-0001') + .send(body); + } finally { + querySpy.mockRestore(); + } + expect(first.status).toBe(503); + expect(first.body.error).toBe('organization_onboarding_retryable'); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + expect((await localAuthorityRows()).operations[0]).toMatchObject({ + workos_organization_id: 'org_onboarding_1', + }); + + invalidateSessionsForUsers([B]); + const retry = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'db-ack-loss-0001') + .send(body); + expect(retry.status, JSON.stringify(retry.body)).toBe(200); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + }); + + it('retries transient finalization lock contention without compensating provider state', async () => { + const blocker = await pool.connect(); + const originalCreateMembership = mocks.createOrganizationMembership.getMockImplementation(); + mocks.createOrganizationMembership.mockImplementationOnce(async (payload: any) => { + const membership = await originalCreateMembership!(payload); + await blocker.query('BEGIN'); + await blocker.query('SELECT workos_user_id FROM users WHERE workos_user_id = $1 FOR UPDATE', [B]); + return membership; + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + let first; + try { + first = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'finalize-lock-retry-0001') + .send(body); + } finally { + await blocker.query('ROLLBACK'); + blocker.release(); + } + expect(first.status).toBe(503); + expect(first.body.error).toBe('organization_onboarding_retryable'); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + expect(mocks.providerOrganizations.size).toBe(1); + expect((await localAuthorityRows()).organizations).toEqual([]); + + invalidateSessionsForUsers([B]); + const retry = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'finalize-lock-retry-0001') + .send(body); + expect(retry.status, JSON.stringify(retry.body)).toBe(200); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.createOrganizationMembership).toHaveBeenCalledTimes(1); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + }); + + it('persists compensation intent before deletion and resumes an ambiguous acknowledgement', async () => { + let reads = 0; + mocks.getUser.mockImplementation(async (id: string) => { + const user = { ...mocks.providerUsers.get(id) }; + if (++reads >= 3) user.email = 'owner@changed-domain.test'; + return user; + }); + const originalQuery = pool.query.bind(pool); + let injected = false; + const querySpy = vi.spyOn(pool as any, 'query').mockImplementation(async (...args: any[]) => { + const result = await (originalQuery as any)(...args); + const sql = typeof args[0] === 'string' ? args[0] : args[0]?.text; + if (!injected && typeof sql === 'string' && sql.includes("SET status = 'compensating'")) { + injected = true; + throw new Error('lost compensation-intent acknowledgement'); + } + return result; + }); + let response; + try { + response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'compensation-ack-loss-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + } finally { + querySpy.mockRestore(); + } + expect(response.status).toBe(503); + expect(response.body.error).toBe('organization_onboarding_reconciliation_required'); + expect(mocks.deleteOrganization).not.toHaveBeenCalled(); + expect((await localAuthorityRows()).operations[0]).toMatchObject({ + status: 'compensating', terminal_outcome: { kind: 'credential_changed' }, + }); + + await pool.query( + `UPDATE organization_onboarding_operations + SET lease_token = NULL, lease_expires_at = NULL + WHERE authenticated_workos_user_id = $1`, + [B], + ); + const result = await reconcilePendingOrganizationOnboarding( + getAuthorizationEnforcementWorkos(), + 1, + ); + expect(result).toMatchObject({ attempted: 1, compensated: 1 }); + expect(mocks.deleteOrganization).toHaveBeenCalledWith('org_onboarding_1'); + expect((await localAuthorityRows()).operations[0]).toMatchObject({ status: 'failed' }); + }); + + it('background reconciliation completes a retryable provider timeout without another client request', async () => { + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + throw Object.assign(new Error('timeout after write'), { code: 'ETIMEDOUT' }); + }); + const response = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'worker-timeout-recovery-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + expect(response.status).toBe(503); + expect(response.body.error).toBe('organization_onboarding_retryable'); + + const result = await reconcilePendingOrganizationOnboarding( + getAuthorizationEnforcementWorkos(), + 1, + ); + expect(result).toMatchObject({ attempted: 1, completed: 1 }); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect((await localAuthorityRows()).organizations).toHaveLength(1); + }); + + it('revisits every manual row without starving behind deduplicated operator escalations', async () => { + const divergentMembership = async ({ userId, organizationId }: any) => ({ + ...exactMembership(userId, organizationId), role: { slug: 'member' }, + }); + mocks.createOrganizationMembership + .mockImplementationOnce(divergentMembership) + .mockImplementationOnce(divergentMembership); + const firstResponse = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'manual-escalation-0001') + .send(createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined })); + const secondResponse = await request(app).post('/api/organizations') + .set('Cookie', cookie(A)) + .set('Idempotency-Key', 'manual-escalation-0002') + .send(createBody({ + organization_name: 'Second Manual Workspace', + is_personal: true, + company_type: undefined, + revenue_tier: undefined, + })); + for (const response of [firstResponse, secondResponse]) { + expect(response.status).toBe(503); + expect(response.body.error).toBe('organization_onboarding_reconciliation_required'); + } + const operations = (await localAuthorityRows()).operations; + expect(operations).toHaveLength(2); + expect(operations).toEqual(expect.arrayContaining([ + expect.objectContaining({ + authenticated_workos_user_id: A, + status: 'manual_reconciliation', + last_error_code: 'provider_state_diverged', + }), + expect.objectContaining({ + authenticated_workos_user_id: B, + status: 'manual_reconciliation', + last_error_code: 'provider_state_diverged', + }), + ])); + + const first = await runOrganizationOnboardingReconciliationJob( + getAuthorizationEnforcementWorkos(), + 1, + 1, + ); + const second = await runOrganizationOnboardingReconciliationJob( + getAuthorizationEnforcementWorkos(), + 1, + 1, + ); + const third = await runOrganizationOnboardingReconciliationJob( + getAuthorizationEnforcementWorkos(), + 1, + 1, + ); + expect(first).toMatchObject({ attempted: 0, manualScanned: 1, manualQueued: 1 }); + expect(second).toMatchObject({ attempted: 0, manualScanned: 1, manualQueued: 1 }); + expect(third).toMatchObject({ attempted: 0, manualScanned: 0, manualQueued: 0 }); + const escalations = await pool.query( + `SELECT category, priority, status, dedup_key, workos_user_id + FROM addie_escalations + WHERE dedup_key = ANY($1) + ORDER BY workos_user_id`, + [operations.map(operation => `organization-onboarding-reconciliation:${operation.id}`)], + ); + expect(escalations.rows).toHaveLength(2); + expect(escalations.rows).toEqual(expect.arrayContaining([ + expect.objectContaining({ + category: 'needs_human_action', priority: 'high', status: 'open', workos_user_id: A, + }), + expect.objectContaining({ + category: 'needs_human_action', priority: 'high', status: 'open', workos_user_id: B, + }), + ])); + }); + + it('compensates an ambiguous provider organization when exact provider identity changes before retry', async () => { + mocks.createOrganization.mockImplementationOnce(async (payload: any) => { + const org = { id: 'org_onboarding_1', name: payload.name, externalId: payload.externalId }; + mocks.providerOrganizations.set(payload.externalId, org); + throw Object.assign(new Error('timeout after write'), { code: 'ETIMEDOUT' }); + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + const first = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'identity-change-retry-0001') + .send(body); + expect(first.status).toBe(503); + expect(first.body.error).toBe('organization_onboarding_retryable'); + + mocks.providerUsers.set(B, { + ...mocks.providerUsers.get(B), email: 'owner@new-provider-domain.test', emailVerified: true, + }); + await pool.query( + 'UPDATE users SET email = $2, email_verified = true WHERE workos_user_id = $1', + [B, 'owner@new-provider-domain.test'], + ); + invalidateSessionsForUsers([B]); + const retry = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'identity-change-retry-0001') + .send(body); + expect(retry.status).toBe(409); + expect(retry.body.error).toBe('onboarding_credential_changed'); + expect(mocks.deleteOrganization).toHaveBeenCalledWith('org_onboarding_1'); + const state = await localAuthorityRows(); + expect(state.organizations).toEqual([]); + expect(state.memberships).toEqual([]); + expect(state.audits).toEqual([]); + expect(state.epochs).toEqual([]); + expect(state.operations).toEqual([expect.objectContaining({ status: 'failed' })]); + }); + + it('recovers an ambiguous owner-membership create without duplicating authority', async () => { + mocks.createOrganizationMembership.mockImplementationOnce(async ({ userId, organizationId }: any) => { + const membership = exactMembership(userId, organizationId); + mocks.providerMemberships.set(membership.id, membership); + throw Object.assign(new Error('timeout after membership write'), { code: 'ETIMEDOUT' }); + }); + const body = createBody({ is_personal: true, company_type: undefined, revenue_tier: undefined }); + const first = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'membership-timeout-0001') + .send(body); + expect(first.status).toBe(503); + expect(first.body.error).toBe('organization_onboarding_retryable'); + expect((await localAuthorityRows()).memberships).toEqual([]); + + const retry = await request(app).post('/api/organizations') + .set('Cookie', cookie()) + .set('Idempotency-Key', 'membership-timeout-0001') + .send(body); + expect(retry.status, JSON.stringify(retry.body)).toBe(200); + expect(mocks.createOrganization).toHaveBeenCalledTimes(1); + expect(mocks.createOrganizationMembership).toHaveBeenCalledTimes(1); + const state = await localAuthorityRows(); + expect(state.memberships).toHaveLength(1); + expect(state.audits).toHaveLength(1); + expect(state.epochs).toHaveLength(1); + }); +}); diff --git a/server/tests/integration/organizations-tier-not-caller-controlled.test.ts b/server/tests/integration/organizations-tier-not-caller-controlled.test.ts index a1ffdbae259..8ff05a60dfe 100644 --- a/server/tests/integration/organizations-tier-not-caller-controlled.test.ts +++ b/server/tests/integration/organizations-tier-not-caller-controlled.test.ts @@ -47,8 +47,8 @@ vi.mock('@workos-inc/node', () => ({ }, })); -vi.mock('../../src/auth/workos-client.js', () => ({ - workos: { +vi.mock('../../src/auth/workos-client.js', () => { + const workos = { userManagement: { createOrganizationMembership: mockCreateOrganizationMembership, listOrganizationMemberships: vi.fn().mockResolvedValue({ data: [] }), @@ -63,8 +63,9 @@ vi.mock('../../src/auth/workos-client.js', () => ({ adminPortal: { generateLink: vi.fn().mockResolvedValue({ link: 'https://test-portal.workos.com' }), }, - }, -})); + }; + return { workos, getAuthorizationEnforcementWorkos: () => workos }; +}); vi.mock('../../src/middleware/auth.js', async (importOriginal) => ({ ...(await importOriginal()), @@ -164,8 +165,8 @@ describe('POST /api/organizations: tier and domain are not caller-controlled', ( membership_tier: 'company_leader', }); - expect(res.status).toBe(403); - expect(res.body.error).toBe('organization_onboarding_disabled'); + expect(res.status).toBe(400); + expect(res.body.error).toBe('agreement_acceptance_required'); expect(mockCreateOrganization).not.toHaveBeenCalled(); expect(mockCreateOrganizationMembership).not.toHaveBeenCalled(); expect((await pool.query('SELECT * FROM organizations WHERE workos_organization_id = $1', [createdOrgId])).rowCount).toBe(0); @@ -184,8 +185,8 @@ describe('POST /api/organizations: tier and domain are not caller-controlled', ( corporate_domain: 'someone-else.example', }); - expect(res.status).toBe(403); - expect(res.body.error).toBe('organization_onboarding_disabled'); + expect(res.status).toBe(400); + expect(res.body.error).toBe('agreement_acceptance_required'); expect(mockCreateOrganization).not.toHaveBeenCalled(); expect(mockCreateOrganizationMembership).not.toHaveBeenCalled(); expect((await pool.query('SELECT * FROM organization_domains WHERE workos_organization_id = $1', [createdOrgId])).rowCount).toBe(0);