diff --git a/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.test.tsx b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.test.tsx new file mode 100644 index 00000000000..7effe3be58c --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.test.tsx @@ -0,0 +1,150 @@ +/** + * @vitest-environment jsdom + */ +import { act, type ReactNode } from 'react' +import { QueryClient, QueryClientProvider, useMutation } from '@tanstack/react-query' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ enabled: false, toggle: vi.fn() })) +vi.mock('next/navigation', () => ({ useParams: () => ({ workspaceId: 'workspace-1' }) })) +vi.mock('@/hooks/queries/inbox', () => ({ + useInboxConfig: () => ({ data: { enabled: mocks.enabled, address: 'inbox@example.com' } }), + useToggleInbox: () => useMutation({ mutationFn: mocks.toggle }), +})) +vi.mock('@sim/emcn', () => ({ + Label: ({ children }: { children: ReactNode }) => {children}, + ChipSwitch: ({ onChange }: { onChange: (value: string) => void }) => ( + <> + + + + ), + ChipModal: ({ open, children }: { open: boolean; children: ReactNode }) => + open ?
{children}
: null, + ChipModalHeader: ({ children }: { children: ReactNode }) =>

{children}

, + ChipModalBody: ({ children }: { children: ReactNode }) =>
{children}
, + ChipModalField: () => null, + ChipModalError: ({ children }: { children: ReactNode }) => + children ?

{children}

: null, + ChipModalFooter: ({ + onCancel, + primaryAction, + }: { + onCancel: () => void + primaryAction: { label: string; onClick: () => void } + }) => ( + <> + + + + ), + ChipConfirmModal: ({ + open, + children, + onOpenChange, + confirm, + }: { + open: boolean + children: ReactNode + onOpenChange: (open: boolean) => void + confirm: { label: string; onClick: () => void } + }) => + open ? ( +
+ {children} + + +
+ ) : null, +})) + +import { InboxEnableToggle } from '@/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle' + +let container: HTMLDivElement +let root: Root +let client: QueryClient +beforeEach(() => { + vi.useFakeTimers() + vi.clearAllMocks() + mocks.enabled = false + ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + container = document.createElement('div') + document.body.appendChild(container) + root = createRoot(container) + client = new QueryClient({ defaultOptions: { mutations: { retry: false } } }) +}) +afterEach(() => { + act(() => root.unmount()) + client.clear() + container.remove() + vi.useRealTimers() +}) + +function render() { + act(() => + root.render( + + + + ) + ) +} +async function click(label: string) { + const button = Array.from(container.querySelectorAll('button')).find( + (button) => button.textContent === label + ) + expect(button).toBeDefined() + await act(async () => { + button?.click() + await vi.runAllTimersAsync() + }) +} + +describe('inbox setup error visibility', () => { + it.each([ + { enabled: false, toggle: 'On', submit: 'Enable' }, + { enabled: true, toggle: 'Off', submit: 'Disable inbox' }, + ])( + 'keeps the dialog open and displays a failed $submit request', + async ({ enabled, toggle, submit }) => { + mocks.enabled = enabled + mocks.toggle.mockRejectedValueOnce(new Error('Email service unavailable')) + render() + await click(toggle) + await click(submit) + expect(container.querySelector('[role="alert"]')?.textContent).toBe( + 'Email service unavailable' + ) + expect(container.querySelector('[role="dialog"]')).not.toBeNull() + await click('Cancel') + await click(toggle) + expect(container.querySelector('[role="alert"]')).toBeNull() + } + ) + + it('clears the failure and closes after a successful retry', async () => { + mocks.toggle + .mockRejectedValueOnce(new Error('Email service unavailable')) + .mockResolvedValueOnce({ enabled: true }) + render() + await click('On') + await click('Enable') + expect(container.querySelector('[role="alert"]')).not.toBeNull() + await click('Enable') + expect(container.querySelector('[role="dialog"]')).toBeNull() + expect(mocks.toggle).toHaveBeenCalledTimes(2) + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.tsx b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.tsx index 0159c6ac807..41c87b24cf0 100644 --- a/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.tsx +++ b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-enable-toggle/inbox-enable-toggle.tsx @@ -1,21 +1,25 @@ 'use client' -import { useCallback, useState } from 'react' +import { useState } from 'react' import { ChipConfirmModal, ChipModal, ChipModalBody, + ChipModalError, ChipModalField, ChipModalFooter, ChipModalHeader, + ChipSwitch, Label, - Switch, } from '@sim/emcn' -import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' import { useParams } from 'next/navigation' import { useInboxConfig, useToggleInbox } from '@/hooks/queries/inbox' -const logger = createLogger('InboxEnableToggle') +const INBOX_OPTIONS = [ + { value: 'enabled', label: 'On' }, + { value: 'disabled', label: 'Off' }, +] as const export function InboxEnableToggle() { const params = useParams() @@ -28,56 +32,68 @@ export function InboxEnableToggle() { const [isDisableOpen, setIsDisableOpen] = useState(false) const [enableUsername, setEnableUsername] = useState('') - const handleToggle = useCallback(async (checked: boolean) => { + function handleToggle(checked: boolean) { + toggleInbox.reset() if (checked) { setIsEnableOpen(true) - return + } else { + setIsDisableOpen(true) } - setIsDisableOpen(true) - }, []) + } - const handleDisable = useCallback(async () => { - try { - await toggleInbox.mutateAsync({ workspaceId, enabled: false }) - setIsDisableOpen(false) - } catch (error) { - logger.error('Failed to disable inbox', { error }) - } - }, [workspaceId, toggleInbox.mutateAsync]) + function handleEnableOpenChange(open: boolean) { + if (!toggleInbox.isPending) setIsEnableOpen(open) + } - const handleEnable = useCallback(async () => { - try { - await toggleInbox.mutateAsync({ - workspaceId, - enabled: true, - username: enableUsername.trim() || undefined, - }) - setIsEnableOpen(false) - setEnableUsername('') - } catch (error) { - logger.error('Failed to enable inbox', { error }) - } - }, [workspaceId, enableUsername, toggleInbox.mutateAsync]) + function handleDisable() { + toggleInbox.mutate( + { workspaceId, enabled: false }, + { onSuccess: () => setIsDisableOpen(false) } + ) + } + + function handleEnable() { + toggleInbox.mutate( + { workspaceId, enabled: true, username: enableUsername.trim() || undefined }, + { + onSuccess: () => { + setIsEnableOpen(false) + setEnableUsername('') + }, + } + ) + } + + const error = toggleInbox.error + ? getErrorMessage(toggleInbox.error, 'Failed to update inbox') + : null return ( <>
- +

Allow this workspace to receive tasks via email

- handleToggle(value === 'enabled')} disabled={toggleInbox.isPending} />
- - setIsEnableOpen(false)}>Enable email inbox + + handleEnableOpenChange(false)}> + Enable email inbox +

An email address will be created for this workspace. Anyone in the allowed senders list @@ -93,11 +109,13 @@ export function InboxEnableToggle() {

Leave blank for an auto-generated address.

+ {error} setIsEnableOpen(false)} + onCancel={() => handleEnableOpenChange(false)} + cancelDisabled={toggleInbox.isPending} primaryAction={{ - label: 'Enable', + label: toggleInbox.isPending ? 'Enabling...' : 'Enable', onClick: handleEnable, disabled: toggleInbox.isPending, }} @@ -125,6 +143,7 @@ export function InboxEnableToggle() {

Your existing conversations and task history will be preserved.

+ {error} ) diff --git a/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-settings-tab/inbox-settings-tab.tsx b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-settings-tab/inbox-settings-tab.tsx index 894a55ef643..b6274cde49d 100644 --- a/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-settings-tab/inbox-settings-tab.tsx +++ b/apps/sim/app/workspace/[workspaceId]/settings/components/inbox/components/inbox-settings-tab/inbox-settings-tab.tsx @@ -296,6 +296,11 @@ export function InboxSettingsTab() { )} + {updateSecretPolicy.error && ( +

+ {getErrorMessage(updateSecretPolicy.error, 'Failed to update secret access')} +

+ )} diff --git a/apps/sim/lib/billing/core/inbox-entitlement.test.ts b/apps/sim/lib/billing/core/inbox-entitlement.test.ts new file mode 100644 index 00000000000..38073f0b69e --- /dev/null +++ b/apps/sim/lib/billing/core/inbox-entitlement.test.ts @@ -0,0 +1,203 @@ +/** + * @vitest-environment node + */ +import { + dbChainMockFns, + resetDbChainMock, + resetEnvFlagsMock, + resetEnvMock, + setEnv, + setEnvFlags, +} from '@sim/testing' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockGetPersonalSubscription, + mockGetOrganizationSubscription, + mockGetWorkspaceWithOwner, + mockGetEffectiveBillingStatus, + mockIsOrganizationBillingBlocked, +} = vi.hoisted(() => ({ + mockGetPersonalSubscription: vi.fn(), + mockGetOrganizationSubscription: vi.fn(), + mockGetWorkspaceWithOwner: vi.fn(), + mockGetEffectiveBillingStatus: vi.fn(), + mockIsOrganizationBillingBlocked: vi.fn(), +})) + +vi.mock('@/lib/billing/core/plan', () => ({ + getHighestPriorityPersonalSubscription: mockGetPersonalSubscription, + getHighestPrioritySubscription: vi.fn(), +})) + +vi.mock('@/lib/billing/core/billing', () => ({ + getOrganizationSubscription: mockGetOrganizationSubscription, +})) + +vi.mock('@/lib/billing/core/access', () => ({ + getEffectiveBillingStatus: mockGetEffectiveBillingStatus, + isOrganizationBillingBlocked: mockIsOrganizationBillingBlocked, +})) + +vi.mock('@/lib/workspaces/permissions/utils', () => ({ + getWorkspaceWithOwner: mockGetWorkspaceWithOwner, +})) + +import { + hasWorkspaceInboxAccess, + hasWorkspaceInboxGraceAccess, +} from '@/lib/billing/core/subscription' + +beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + setEnv({ COPILOT_API_KEY: 'test-copilot-key' }) + setEnvFlags({ isHosted: true, isBillingEnabled: true, isInboxEnabled: false }) + mockGetWorkspaceWithOwner.mockResolvedValue({ + id: 'workspace-1', + billedAccountUserId: 'payer-1', + organizationId: null, + }) + mockGetPersonalSubscription.mockResolvedValue(null) + mockGetOrganizationSubscription.mockResolvedValue(null) + mockGetEffectiveBillingStatus.mockResolvedValue({ + billingBlocked: false, + billingBlockedReason: null, + blockedByOrgOwner: false, + }) + mockIsOrganizationBillingBlocked.mockResolvedValue(false) +}) + +afterEach(() => { + resetEnvFlagsMock() + resetEnvMock() +}) + +describe('Sim Mailer hosted entitlement', () => { + it.each([ + { isInboxEnabled: true, isBillingEnabled: true }, + { isInboxEnabled: false, isBillingEnabled: false }, + { isInboxEnabled: true, isBillingEnabled: false }, + ])('requires a qualifying payer despite deployment flags %o', async (flags) => { + setEnvFlags(flags) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(false) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(false) + expect(mockGetPersonalSubscription).toHaveBeenCalledWith('payer-1') + }) + + it.each([ + ['pro_25000', true], + ['enterprise', true], + ['pro_6000', false], + ['pro', false], + ['free', false], + ])('checks the personal workspace payer plan %s', async (plan, expected) => { + mockGetPersonalSubscription.mockResolvedValue({ plan, status: 'active' }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(expected) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(expected) + }) + + it.each([ + ['team_25000', true], + ['enterprise', true], + ['team_6000', false], + ])( + 'checks the organization payer plan %s without requiring a personal plan', + async (plan, expected) => { + mockGetWorkspaceWithOwner.mockResolvedValue({ + id: 'workspace-1', + billedAccountUserId: 'payer-1', + organizationId: 'org-1', + }) + dbChainMockFns.limit.mockResolvedValueOnce([{ plan, status: 'active' }]) + mockGetOrganizationSubscription.mockResolvedValue({ plan, status: 'active' }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(expected) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(expected) + expect(mockGetPersonalSubscription).not.toHaveBeenCalled() + expect(mockGetOrganizationSubscription).toHaveBeenCalledWith('org-1', { onError: 'throw' }) + } + ) + + it('blocks a past-due Max payer from use while preserving provisioned resources', async () => { + mockGetPersonalSubscription.mockResolvedValue({ plan: 'pro_25000', status: 'past_due' }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(false) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) + + it('blocks a billing-blocked Max payer from use while preserving provisioned resources', async () => { + mockGetPersonalSubscription.mockResolvedValue({ plan: 'pro_25000', status: 'active' }) + mockGetEffectiveBillingStatus.mockResolvedValue({ billingBlocked: true }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(false) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) + + it('requires the execution key for use without destroying resources when it is missing', async () => { + setEnv({ COPILOT_API_KEY: undefined }) + mockGetPersonalSubscription.mockResolvedValue({ plan: 'pro_25000', status: 'active' }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(false) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) +}) + +describe('Sim Mailer cleanup uncertainty', () => { + it('preserves the inbox if the workspace cannot be found', async () => { + mockGetWorkspaceWithOwner.mockResolvedValue(null) + + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) + + it('preserves the inbox on a workspace lookup failure', async () => { + mockGetWorkspaceWithOwner.mockRejectedValue(new Error('Database unavailable')) + + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) + + it('requires the personal subscription reader to surface errors and preserves the inbox', async () => { + mockGetPersonalSubscription.mockRejectedValue(new Error('Database unavailable')) + + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + expect(mockGetPersonalSubscription).toHaveBeenCalledWith('payer-1', { onError: 'throw' }) + }) + + it('requires the organization subscription reader to surface errors and preserves the inbox', async () => { + mockGetWorkspaceWithOwner.mockResolvedValue({ + id: 'workspace-1', + billedAccountUserId: 'payer-1', + organizationId: 'org-1', + }) + mockGetOrganizationSubscription.mockRejectedValue(new Error('Database unavailable')) + + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + expect(mockGetOrganizationSubscription).toHaveBeenCalledWith('org-1', { onError: 'throw' }) + }) + + it('retains past-due Max for Teams resources', async () => { + mockGetWorkspaceWithOwner.mockResolvedValue({ + id: 'workspace-1', + billedAccountUserId: 'payer-1', + organizationId: 'org-1', + }) + mockGetOrganizationSubscription.mockResolvedValue({ plan: 'team_25000', status: 'past_due' }) + + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + }) +}) + +describe('Sim Mailer self-hosted overrides', () => { + it.each([ + { isInboxEnabled: true, isBillingEnabled: true }, + { isInboxEnabled: false, isBillingEnabled: false }, + ])('preserves self-hosted configuration %o', async (flags) => { + setEnvFlags({ isHosted: false, ...flags }) + + await expect(hasWorkspaceInboxAccess('workspace-1')).resolves.toBe(true) + await expect(hasWorkspaceInboxGraceAccess('workspace-1')).resolves.toBe(true) + expect(mockGetWorkspaceWithOwner).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/billing/core/subscription.ts b/apps/sim/lib/billing/core/subscription.ts index 10b709b54bd..64c6b0c8f6d 100644 --- a/apps/sim/lib/billing/core/subscription.ts +++ b/apps/sim/lib/billing/core/subscription.ts @@ -803,16 +803,14 @@ const hasMaxTierWorkspaceAccess = cache( * Inbox. * * Otherwise returns true if: - * - INBOX_ENABLED env var is set (self-hosted override), OR - * - billing is disabled, OR + * - on self-hosted deployments, INBOX_ENABLED is set or billing is disabled, OR * - the workspace belongs to an organization on a Max/enterprise plan (org-mode), OR * - the billed user has an individual Max/enterprise subscription (personal workspace). */ export async function hasWorkspaceInboxAccess(workspaceId: string): Promise { try { if (!env.COPILOT_API_KEY) return false - if (isInboxEnabled) return true - if (!isBillingEnabled) return true + if (!isHosted && (isInboxEnabled || !isBillingEnabled)) return true return await hasMaxTierWorkspaceAccess(workspaceId) } catch (error) { logger.error('Error checking workspace inbox access', { error, workspaceId }) @@ -834,12 +832,12 @@ export async function hasWorkspaceInboxAccess(workspaceId: string): Promise { try { - if (isInboxEnabled) return true - if (!isBillingEnabled) return true + if (!isHosted && (isInboxEnabled || !isBillingEnabled)) return true return await hasWorkspaceTierAccess(workspaceId, isMaxTier, { intent: 'retention', onMissingWorkspace: true, + onError: 'throw', }) } catch (error) { logger.error('Error checking workspace inbox grace access', { error, workspaceId }) diff --git a/apps/sim/lib/core/outbox/processor.test.ts b/apps/sim/lib/core/outbox/processor.test.ts index e63b025eca5..d20af17bc58 100644 --- a/apps/sim/lib/core/outbox/processor.test.ts +++ b/apps/sim/lib/core/outbox/processor.test.ts @@ -33,6 +33,7 @@ vi.mock('@/lib/knowledge/application/slack-search/outbox', () => ({ vi.mock('@/lib/knowledge/documents/processing-outbox-handler', () => ({ knowledgeDocumentProcessingOutboxHandlers: {}, })) +vi.mock('@/lib/mothership/inbox/cleanup-outbox', () => ({ inboxCleanupOutboxHandlers: {} })) vi.mock('@/lib/organizations/resource-cleanup', () => ({ organizationResourceCleanupOutboxHandlers: {}, })) diff --git a/apps/sim/lib/core/outbox/processor.ts b/apps/sim/lib/core/outbox/processor.ts index 7c9fbe9fea7..464460579bc 100644 --- a/apps/sim/lib/core/outbox/processor.ts +++ b/apps/sim/lib/core/outbox/processor.ts @@ -18,6 +18,7 @@ import { slackSearchOutboxHandlers } from '@/lib/knowledge/application/slack-sea import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error' import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler' import { recoverKnowledgeDocumentProcessing } from '@/lib/knowledge/documents/processing-recovery' +import { inboxCleanupOutboxHandlers } from '@/lib/mothership/inbox/cleanup-outbox' import { organizationResourceCleanupOutboxHandlers } from '@/lib/organizations/resource-cleanup' import { permissionAccessRequestOutboxHandlers } from '@/lib/permission-access-requests/notifications' import { workspaceFileLiveDocOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-live-doc-outbox' @@ -42,6 +43,7 @@ const handlers = { ...directGrantOutboxHandlers, ...knowledgeDocumentProcessingOutboxHandlers, ...organizationResourceCleanupOutboxHandlers, + ...inboxCleanupOutboxHandlers, ...permissionAccessRequestOutboxHandlers, ...workspaceFileLiveDocOutboxHandlers, ...workspaceFileStorageCleanupOutboxHandlers, diff --git a/apps/sim/lib/core/outbox/service.ts b/apps/sim/lib/core/outbox/service.ts index 967e2a6e487..f3b857699e4 100644 --- a/apps/sim/lib/core/outbox/service.ts +++ b/apps/sim/lib/core/outbox/service.ts @@ -83,7 +83,8 @@ export interface DeferredOutboxHandlerResult { * Defaults to true for an external acknowledgement with a finite retry * budget. False is reserved for waits on an internal dependency whose own * outbox row independently reaches completed or dead-letter, and for - * bounded continuation after durable progress (`continueOutboxHandler`). + * bounded continuation after durable progress (`continueOutboxHandler`), + * or external polling with a separately persisted, finite poll allowance. */ consumeAttempt?: boolean } diff --git a/apps/sim/lib/mothership/inbox/agentmail-client.test.ts b/apps/sim/lib/mothership/inbox/agentmail-client.test.ts new file mode 100644 index 00000000000..17a5bc3790c --- /dev/null +++ b/apps/sim/lib/mothership/inbox/agentmail-client.test.ts @@ -0,0 +1,68 @@ +/** + * @vitest-environment node + */ +import { resetEnvMock, setEnv } from '@sim/testing' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + createInbox, + createWebhook, + deleteInbox, + deleteWebhook, + getInbox, +} from '@/lib/mothership/inbox/agentmail-client' + +const fetchMock = vi.fn() +beforeEach(() => { + fetchMock.mockReset() + setEnv({ AGENTMAIL_API_KEY: 'test-key' }) + vi.stubGlobal('fetch', fetchMock) +}) +afterEach(() => { + vi.unstubAllGlobals() + resetEnvMock() +}) + +describe('AgentMail resource deletion', () => { + it.each([deleteInbox, deleteWebhook])( + 'accepts an empty 202 without attempting JSON parsing', + async (remove) => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 202 })) + await expect(remove('resource-id')).resolves.toBe(false) + } + ) + it.each([204, 404])('treats status %s as completed deletion', async (status) => { + fetchMock.mockResolvedValueOnce(new Response(null, { status })) + await expect(deleteInbox('resource-id')).resolves.toBe(true) + }) + it('does not treat a conflict as completed deletion', async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 409 })) + await expect(deleteInbox('resource-id')).rejects.toThrow() + }) + it('returns null only for a missing inbox', async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 404 })) + await expect(getInbox('resource-id')).resolves.toBeNull() + fetchMock.mockResolvedValueOnce(new Response(null, { status: 503 })) + await expect(getInbox('resource-id')).rejects.toThrow() + }) + it('never exposes provider response bodies in user-facing errors', async () => { + fetchMock.mockResolvedValueOnce(new Response('private provider diagnostic', { status: 400 })) + await expect(createInbox({ username: 'test' })).rejects.toThrow('Check the email prefix') + }) + + it('reports an address conflict only when creating an inbox', async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 409 })) + await expect(createInbox({ username: 'test' })).rejects.toThrow( + 'This email address is unavailable' + ) + }) + it('reports webhook conflicts as a service failure instead of an address conflict', async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 409 })) + await expect( + createWebhook({ + url: 'https://example.com/webhook', + eventTypes: ['message.received'], + inboxIds: ['inbox@example.com'], + }) + ).rejects.toThrow('The email service is unavailable') + }) +}) diff --git a/apps/sim/lib/mothership/inbox/agentmail-client.ts b/apps/sim/lib/mothership/inbox/agentmail-client.ts index 3f7a0dfe5e4..9373eb7ce01 100644 --- a/apps/sim/lib/mothership/inbox/agentmail-client.ts +++ b/apps/sim/lib/mothership/inbox/agentmail-client.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { env } from '@/lib/core/config/env' +import { readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits' import type { AgentMailAttachment, AgentMailInbox, @@ -20,7 +21,23 @@ function getApiKey(): string { return key } -async function request(path: string, options: RequestInit = {}): Promise { +class AgentMailError extends Error { + constructor( + readonly status: number, + path: string + ) { + super( + status === 409 && path === '/inboxes' + ? 'This email address is unavailable. Choose another prefix or try again later.' + : status === 400 && path === '/inboxes' + ? 'Unable to create this inbox. Check the email prefix and try again, or contact support.' + : 'The email service is unavailable. Please try again later or contact support.' + ) + this.name = 'AgentMailError' + } +} + +async function requestResponse(path: string, options: RequestInit = {}): Promise { const url = `${BASE_URL}${path}` const response = await fetch(url, { ...options, @@ -32,20 +49,59 @@ async function request(path: string, options: RequestInit = {}): Promise { }) if (!response.ok) { - const body = await response.text().catch(() => '') - logger.error('AgentMail API error', { - status: response.status, - path, - body, - }) - throw new Error(`AgentMail API error: ${response.status} ${body}`) + logger.error('AgentMail API error', { status: response.status, path }) + await response.body?.cancel() + throw new AgentMailError(response.status, path) + } + return response +} + +async function request(path: string, options: RequestInit = {}): Promise { + const response = await requestResponse(path, options) + return response.json() as Promise +} + +function withResourceTimeout(options: RequestInit): RequestInit { + return { + ...options, + signal: options.signal + ? AbortSignal.any([options.signal, AbortSignal.timeout(15_000)]) + : AbortSignal.timeout(15_000), } +} + +async function requestResource(path: string, options: RequestInit = {}): Promise { + const response = await requestResponse(path, withResourceTimeout(options)) + return readResponseJsonWithLimit(response, { + maxBytes: 64 * 1024, + label: 'AgentMail resource', + }) +} - if (response.status === 204) { - return undefined as T +/** Treat already-absent resources as deleted, and distinguish asynchronous acceptance. */ +async function deleteResource(path: string, signal?: AbortSignal): Promise { + try { + const response = await requestResponse(path, withResourceTimeout({ method: 'DELETE', signal })) + await response.body?.cancel() + return response.status !== 202 + } catch (error) { + if (error instanceof AgentMailError && error.status === 404) return true + throw error } +} - return response.json() as Promise +export async function getInbox( + inboxId: string, + signal?: AbortSignal +): Promise { + try { + return await requestResource(`/inboxes/${encodeURIComponent(inboxId)}`, { + signal, + }) + } catch (error) { + if (error instanceof AgentMailError && error.status === 404) return null + throw error + } } export async function createInbox(opts: { @@ -53,7 +109,7 @@ export async function createInbox(opts: { displayName?: string }): Promise { const domain = env.AGENTMAIL_DOMAIN - return request('/inboxes', { + return requestResource('/inboxes', { method: 'POST', body: JSON.stringify({ username: opts.username, @@ -63,10 +119,8 @@ export async function createInbox(opts: { }) } -export async function deleteInbox(inboxId: string): Promise { - return request(`/inboxes/${encodeURIComponent(inboxId)}`, { - method: 'DELETE', - }) +export function deleteInbox(inboxId: string, signal?: AbortSignal): Promise { + return deleteResource(`/inboxes/${encodeURIComponent(inboxId)}`, signal) } export async function createWebhook(opts: { @@ -74,7 +128,7 @@ export async function createWebhook(opts: { eventTypes: string[] inboxIds: string[] }): Promise { - return request('/webhooks', { + return requestResource('/webhooks', { method: 'POST', body: JSON.stringify({ url: opts.url, @@ -84,10 +138,8 @@ export async function createWebhook(opts: { }) } -export async function deleteWebhook(webhookId: string): Promise { - return request(`/webhooks/${encodeURIComponent(webhookId)}`, { - method: 'DELETE', - }) +export function deleteWebhook(webhookId: string, signal?: AbortSignal): Promise { + return deleteResource(`/webhooks/${encodeURIComponent(webhookId)}`, signal) } export async function replyToMessage( diff --git a/apps/sim/lib/mothership/inbox/cleanup-outbox.test.ts b/apps/sim/lib/mothership/inbox/cleanup-outbox.test.ts new file mode 100644 index 00000000000..0d2939ef6f1 --- /dev/null +++ b/apps/sim/lib/mothership/inbox/cleanup-outbox.test.ts @@ -0,0 +1,231 @@ +/** + * @vitest-environment node + */ + +import { db } from '@sim/db' +import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { OutboxEventContext } from '@/lib/core/outbox/service' +import * as outboxService from '@/lib/core/outbox/service' + +const mocks = vi.hoisted(() => ({ + getInbox: vi.fn(), + deleteInbox: vi.fn(), + deleteWebhook: vi.fn(), +})) +vi.mock('@/lib/mothership/inbox/agentmail-client', () => mocks) + +import { + cancelInboxCleanup, + type InboxCleanupPayload, + inboxCleanupOutboxHandlers, + processInboxCleanupNow, +} from '@/lib/mothership/inbox/cleanup-outbox' + +const cleanup = inboxCleanupOutboxHandlers['inbox.resources.cleanup'] +const payload = { + inboxId: 'old@example.com', + inboxCreatedAt: '2025-01-01T00:00:00Z', + webhookId: 'old-hook', +} +let context: OutboxEventContext + +beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mocks.getInbox.mockResolvedValue({ + inbox_id: payload.inboxId, + created_at: payload.inboxCreatedAt, + }) + mocks.deleteInbox.mockResolvedValue(true) + mocks.deleteWebhook.mockResolvedValue(true) + context = { + eventId: 'event-1', + eventType: 'inbox.resources.cleanup', + attempts: 0, + maxAttempts: 10, + signal: new AbortController().signal, + checkpointPayload: vi.fn().mockResolvedValue(undefined), + } +}) + +describe('durable inbox cleanup', () => { + it('refuses resources still referenced by an active configuration', async () => { + queueTableRows(schemaMock.mothershipInboxWebhook, [{ id: 'active-hook' }]) + await expect(cleanup(payload, context)).rejects.toThrow('still in use') + expect(mocks.deleteWebhook).not.toHaveBeenCalled() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('refuses an inbox still referenced by a workspace', async () => { + queueTableRows(schemaMock.workspace, [{ id: 'workspace-1' }]) + await expect(cleanup({ ...payload, webhookId: null }, context)).rejects.toThrow('still in use') + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('surfaces provider failures for the outbox retry policy', async () => { + mocks.deleteWebhook.mockRejectedValueOnce(new Error('Service unavailable')) + await expect(cleanup(payload, context)).rejects.toThrow('Service unavailable') + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('checkpoints accepted deletion and polls on retry instead of deleting again', async () => { + mocks.deleteInbox.mockResolvedValueOnce(false) + await expect(cleanup(payload, context)).resolves.toMatchObject({ outcome: 'deferred' }) + expect(context.checkpointPayload).toHaveBeenCalledWith({ inboxDeleteAccepted: true }) + await expect( + cleanup({ ...payload, inboxDeleteAccepted: true }, context) + ).resolves.toMatchObject({ outcome: 'deferred' }) + expect(mocks.deleteInbox).toHaveBeenCalledTimes(1) + mocks.getInbox.mockResolvedValueOnce(null) + await expect( + cleanup({ ...payload, inboxDeleteAccepted: true }, context) + ).resolves.toBeUndefined() + }) + + it('does not delete a new inbox that reused the retired address', async () => { + mocks.getInbox.mockResolvedValueOnce({ + inbox_id: payload.inboxId, + created_at: '2025-02-01T00:00:00Z', + }) + await expect(cleanup(payload, context)).resolves.toBeUndefined() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('completes already-absent resources without issuing another inbox delete', async () => { + mocks.getInbox.mockResolvedValueOnce(null) + await expect(cleanup(payload, context)).resolves.toBeUndefined() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('does not lose asynchronous webhook deletion', async () => { + mocks.deleteWebhook.mockResolvedValueOnce(false) + await expect(cleanup(payload, context)).resolves.toMatchObject({ outcome: 'deferred' }) + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('honors cancellation before provider side effects', async () => { + context.signal = AbortSignal.abort() + await expect(cleanup(payload, context)).rejects.toThrow() + expect(mocks.deleteWebhook).not.toHaveBeenCalled() + }) + + it('requires a creation timestamp before deleting an address', async () => { + await expect(cleanup({ ...payload, inboxCreatedAt: null }, context)).rejects.toThrow() + expect(mocks.deleteWebhook).not.toHaveBeenCalled() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('records that cleanup started before deleting resources so a retry cannot be activated', async () => { + await cleanup(payload, context) + expect(context.checkpointPayload).toHaveBeenCalledWith({ cleanupStarted: true }) + expect(vi.mocked(context.checkpointPayload).mock.invocationCallOrder[0]).toBeLessThan( + mocks.deleteWebhook.mock.invocationCallOrder[0] + ) + }) + + it('does not delete anything if the cleanup lease cannot checkpoint its start', async () => { + vi.mocked(context.checkpointPayload).mockRejectedValueOnce(new Error('Lease lost')) + await expect(cleanup(payload, context)).rejects.toThrow('Lease lost') + expect(mocks.deleteWebhook).not.toHaveBeenCalled() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('refuses activation when rollback is no longer cancelable', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([]) + await expect(cancelInboxCleanup(db, 'event-1')).rejects.toThrow('Inbox setup expired') + }) + + it.each(['webhook', 'inbox'])( + 'allows more than ten expected %s polls without spending failure attempts', + async (resource) => { + const pending: InboxCleanupPayload = { ...payload } + vi.mocked(context.checkpointPayload).mockImplementation(async (patch) => { + Object.assign(pending, patch) + }) + if (resource === 'webhook') mocks.deleteWebhook.mockResolvedValue(false) + else mocks.deleteInbox.mockResolvedValue(false) + + for (let poll = 0; poll < 11; poll++) { + await expect(cleanup(pending, context)).resolves.toMatchObject({ + outcome: 'deferred', + consumeAttempt: false, + minimumBackoffMs: 30_000, + }) + } + expect(pending.deletionPollsRemaining).toBe(109) + mocks.deleteWebhook.mockResolvedValue(true) + mocks.getInbox.mockResolvedValueOnce(null) + await expect(cleanup(pending, context)).resolves.toBeUndefined() + } + ) + + it.each(['webhook', 'inbox'])( + 'bounds perpetual %s polling and returns to the failure budget', + async (resource) => { + const pending: InboxCleanupPayload = { ...payload, deletionPollsRemaining: 1 } + vi.mocked(context.checkpointPayload).mockImplementation(async (patch) => { + Object.assign(pending, patch) + }) + if (resource === 'webhook') mocks.deleteWebhook.mockResolvedValue(false) + else mocks.deleteInbox.mockResolvedValue(false) + + await expect(cleanup(pending, context)).resolves.toMatchObject({ consumeAttempt: false }) + expect(pending.deletionPollsRemaining).toBe(0) + const exhausted = await cleanup(pending, context) + expect(exhausted).toMatchObject({ + outcome: 'deferred', + reason: expect.stringContaining('exhausted'), + }) + expect(exhausted).not.toHaveProperty('consumeAttempt', false) + expect(pending.deletionPollsRemaining).toBe(0) + } + ) + + it('continues charging actual provider failures to the outbox error budget', async () => { + mocks.deleteWebhook.mockRejectedValueOnce(new Error('Connection failed')) + await expect(cleanup({ ...payload, deletionPollsRemaining: 100 }, context)).rejects.toThrow( + 'Connection failed' + ) + expect(context.checkpointPayload).not.toHaveBeenCalledWith( + expect.objectContaining({ deletionPollsRemaining: expect.any(Number) }) + ) + }) +}) + +describe('immediate rollback processing', () => { + afterEach(() => vi.restoreAllMocks()) + + it('makes failed activation rollback due before asking the outbox to claim it', async () => { + const process = vi.spyOn(outboxService, 'processOutboxEventById').mockResolvedValue('completed') + const before = Date.now() + await processInboxCleanupNow('rollback-event', { expedite: true }) + expect(dbChainMockFns.update).toHaveBeenCalledWith(schemaMock.outboxEvent) + const scheduled = dbChainMockFns.set.mock.calls[0][0].availableAt + expect(scheduled.getTime()).toBeGreaterThanOrEqual(before) + expect(scheduled.getTime()).toBeLessThanOrEqual(Date.now()) + expect(process).toHaveBeenCalledWith('rollback-event', inboxCleanupOutboxHandlers) + expect(dbChainMockFns.set.mock.invocationCallOrder[0]).toBeLessThan( + process.mock.invocationCallOrder[0] + ) + }) + + it('leaves the durable fallback intact if the database cannot expedite rollback', async () => { + const process = vi.spyOn(outboxService, 'processOutboxEventById').mockResolvedValue('completed') + dbChainMockFns.update.mockImplementationOnce(() => { + throw new Error('Database unavailable') + }) + await expect( + processInboxCleanupNow('rollback-event', { expedite: true }) + ).resolves.toBeUndefined() + expect(process).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) + + it('does not change availability for ordinary cleanup or retries', async () => { + const process = vi.spyOn(outboxService, 'processOutboxEventById').mockResolvedValue('pending') + await processInboxCleanupNow('cleanup-event') + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(process).toHaveBeenCalledWith('cleanup-event', inboxCleanupOutboxHandlers) + }) +}) diff --git a/apps/sim/lib/mothership/inbox/cleanup-outbox.ts b/apps/sim/lib/mothership/inbox/cleanup-outbox.ts new file mode 100644 index 00000000000..0c37df5315c --- /dev/null +++ b/apps/sim/lib/mothership/inbox/cleanup-outbox.ts @@ -0,0 +1,146 @@ +import { db } from '@sim/db' +import { mothershipInboxWebhook, outboxEvent, workspace } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { and, eq, gt, sql } from 'drizzle-orm' +import { z } from 'zod' +import { + deferOutboxHandler, + enqueueOutboxEvent, + type OutboxEventContext, + type OutboxHandler, + type OutboxHandlerRegistry, + processOutboxEventById, +} from '@/lib/core/outbox/service' +import * as agentmail from '@/lib/mothership/inbox/agentmail-client' + +const logger = createLogger('InboxCleanup') +const INBOX_CLEANUP_EVENT = 'inbox.resources.cleanup' +const MAX_DELETION_POLLS = 120 +const DELETION_POLL_INTERVAL_MS = 30_000 + +const cleanupPayloadSchema = z + .object({ + inboxId: z.string().min(1).max(320).nullable(), + inboxCreatedAt: z.string().datetime({ offset: true }).nullable(), + webhookId: z.string().min(1).max(256).nullable(), + inboxDeleteAccepted: z.boolean().optional(), + cleanupStarted: z.boolean().optional(), + deletionPollsRemaining: z.number().int().min(0).max(MAX_DELETION_POLLS).optional(), + }) + .refine((payload) => !payload.inboxId || payload.inboxCreatedAt !== null) + +export type InboxCleanupPayload = z.infer + +/** Expected provider waits have a separate finite allowance from failures such as network errors. */ +async function waitForDeletion( + payload: InboxCleanupPayload, + context: OutboxEventContext, + reason: string +) { + const remaining = payload.deletionPollsRemaining ?? MAX_DELETION_POLLS + if (remaining === 0) { + return deferOutboxHandler(`${reason}: polling allowance exhausted`, DELETION_POLL_INTERVAL_MS) + } + await context.checkpointPayload({ deletionPollsRemaining: remaining - 1 }) + return deferOutboxHandler(reason, DELETION_POLL_INTERVAL_MS, false) +} + +const cleanupInboxResources: OutboxHandler = async (rawPayload, context) => { + const payload = cleanupPayloadSchema.parse(rawPayload) + context.signal.throwIfAborted() + /** Retries and recovered leases become pending again, but must never be activated afterward. */ + if (!payload.cleanupStarted) await context.checkpointPayload({ cleanupStarted: true }) + + if (payload.webhookId) { + const [active] = await db + .select({ id: mothershipInboxWebhook.id }) + .from(mothershipInboxWebhook) + .where(eq(mothershipInboxWebhook.webhookId, payload.webhookId)) + .limit(1) + if (active) throw new Error('Inbox webhook is still in use') + if (!(await agentmail.deleteWebhook(payload.webhookId, context.signal))) { + return waitForDeletion(payload, context, 'Waiting for webhook deletion') + } + } + + if (!payload.inboxId) return + context.signal.throwIfAborted() + const [active] = await db + .select({ id: workspace.id }) + .from(workspace) + .where(eq(workspace.inboxProviderId, payload.inboxId)) + .limit(1) + if (active) throw new Error('Inbox is still in use') + + const inbox = await agentmail.getInbox(payload.inboxId, context.signal) + /** Email addresses can be reused; a retry must never delete a replacement inbox. */ + if (!inbox || inbox.created_at !== payload.inboxCreatedAt) return + if (!payload.inboxDeleteAccepted) { + context.signal.throwIfAborted() + if (await agentmail.deleteInbox(payload.inboxId, context.signal)) return + await context.checkpointPayload({ inboxDeleteAccepted: true }) + } + return waitForDeletion(payload, context, 'Waiting for inbox deletion') +} + +export const inboxCleanupOutboxHandlers = { + [INBOX_CLEANUP_EVENT]: cleanupInboxResources, +} satisfies OutboxHandlerRegistry + +/** Retains resource identities in the same transaction that removes their active configuration. */ +export function enqueueInboxCleanup( + executor: Pick, + payload: InboxCleanupPayload, + availableAt?: Date +): Promise { + return enqueueOutboxEvent(executor, INBOX_CLEANUP_EVENT, cleanupPayloadSchema.parse(payload), { + availableAt, + }) +} + +/** Attempts committed cleanup immediately, leaving failures to the durable retry worker. */ +export async function processInboxCleanupNow( + eventId: string, + options: { expedite?: boolean } = {} +): Promise { + try { + if (options.expedite) { + const now = new Date() + /** Only release the activation grace delay; preserve claimed work and retry backoff. */ + await db + .update(outboxEvent) + .set({ availableAt: now }) + .where( + and( + eq(outboxEvent.id, eventId), + eq(outboxEvent.eventType, INBOX_CLEANUP_EVENT), + eq(outboxEvent.status, 'pending'), + gt(outboxEvent.availableAt, now), + sql`${outboxEvent.payload}->>'cleanupStarted' IS DISTINCT FROM 'true'` + ) + ) + } + await processOutboxEventById(eventId, inboxCleanupOutboxHandlers) + } catch (error) { + logger.warn('Inbox cleanup remains queued', { eventId, error }) + } +} + +/** Cancels unclaimed rollback only inside the transaction that activates its resources. */ +export async function cancelInboxCleanup( + executor: Pick, + eventId: string +): Promise { + const [canceled] = await executor + .delete(outboxEvent) + .where( + and( + eq(outboxEvent.id, eventId), + eq(outboxEvent.eventType, INBOX_CLEANUP_EVENT), + eq(outboxEvent.status, 'pending'), + sql`${outboxEvent.payload}->>'cleanupStarted' IS DISTINCT FROM 'true'` + ) + ) + .returning({ id: outboxEvent.id }) + if (!canceled) throw new Error('Inbox setup expired. Please try again.') +} diff --git a/apps/sim/lib/mothership/inbox/lifecycle.test.ts b/apps/sim/lib/mothership/inbox/lifecycle.test.ts new file mode 100644 index 00000000000..8b055fb33c3 --- /dev/null +++ b/apps/sim/lib/mothership/inbox/lifecycle.test.ts @@ -0,0 +1,230 @@ +/** + * @vitest-environment node + */ +import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + createInbox: vi.fn(), + createWebhook: vi.fn(), + getInbox: vi.fn(), + enqueue: vi.fn(), + process: vi.fn(), + cancel: vi.fn(), + deleteInbox: vi.fn(), + deleteWebhook: vi.fn(), +})) +vi.mock('@/lib/mothership/inbox/agentmail-client', () => mocks) +vi.mock('@/lib/mothership/inbox/cleanup-outbox', () => ({ + cancelInboxCleanup: mocks.cancel, + enqueueInboxCleanup: mocks.enqueue, + processInboxCleanupNow: mocks.process, +})) +vi.mock('@/lib/core/utils/urls', () => ({ getBaseUrl: () => 'https://example.com' })) + +import { disableInbox, enableInbox, updateInboxAddress } from '@/lib/mothership/inbox/lifecycle' + +const oldState = { + enabled: true, + address: 'old@example.com', + providerId: 'old@example.com', + webhookId: 'old-hook', +} +const emptyState = { enabled: false, address: null, providerId: null, webhookId: null } +const createdAt = '2025-01-01T00:00:00Z' +const newInbox = { inbox_id: 'new@example.com', created_at: createdAt } + +beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mocks.createInbox.mockResolvedValue(newInbox) + mocks.createWebhook.mockResolvedValue({ webhook_id: 'new-hook', secret: 'test-secret' }) + mocks.getInbox.mockResolvedValue({ inbox_id: oldState.providerId, created_at: createdAt }) + mocks.enqueue.mockResolvedValue('cleanup-event') + mocks.process.mockResolvedValue(undefined) + mocks.cancel.mockResolvedValue(undefined) + mocks.deleteInbox.mockResolvedValue(true) + mocks.deleteWebhook.mockResolvedValue(true) +}) + +function queueState(state = oldState) { + queueTableRows(schemaMock.workspace, [state]) +} +function queueLockedState(state = oldState) { + queueTableRows(schemaMock.workspace, [{ id: 'workspace-1' }]) + queueState(state) +} + +describe('inbox lifecycle failure safety', () => { + it('keeps the existing inbox when a replacement address is unavailable', async () => { + queueState() + mocks.createInbox.mockRejectedValueOnce(new Error('Address unavailable')) + await expect(updateInboxAddress('workspace-1', 'new')).rejects.toThrow('Address unavailable') + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(mocks.enqueue).not.toHaveBeenCalled() + }) + + it('retains the old configuration and cleans up only the new inbox when webhook creation fails', async () => { + queueState() + mocks.createWebhook.mockRejectedValueOnce(new Error('Webhook limit')) + await expect(updateInboxAddress('workspace-1', 'new')).rejects.toThrow('Webhook limit') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mocks.enqueue).toHaveBeenCalledWith(expect.anything(), { + inboxId: newInbox.inbox_id, + inboxCreatedAt: createdAt, + webhookId: null, + }) + }) + + it('commits the replacement and cleanup identities together before deleting old resources', async () => { + queueState() + queueLockedState() + await expect(updateInboxAddress('workspace-1', 'new')).resolves.toMatchObject({ + address: newInbox.inbox_id, + }) + expect(mocks.enqueue).toHaveBeenCalledWith(expect.anything(), { + inboxId: oldState.providerId, + inboxCreatedAt: createdAt, + webhookId: 'old-hook', + }) + expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(1) + expect(dbChainMockFns.for).toHaveBeenCalledWith('update') + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ inboxEnabled: true, inboxProviderId: newInbox.inbox_id }) + ) + expect(mocks.process.mock.invocationCallOrder[0]).toBeGreaterThan( + dbChainMockFns.set.mock.invocationCallOrder[0] + ) + }) + + it('does not overwrite a concurrent change and rolls back only its own resources', async () => { + queueState() + queueLockedState({ ...oldState, webhookId: 'concurrent-hook' }) + await expect(updateInboxAddress('workspace-1', 'new')).rejects.toThrow('Inbox settings changed') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mocks.enqueue).toHaveBeenCalledWith( + expect.anything(), + { + inboxId: newInbox.inbox_id, + inboxCreatedAt: createdAt, + webhookId: 'new-hook', + }, + expect.any(Date) + ) + }) + + it('does not clear configuration when it cannot durably enqueue cleanup', async () => { + queueState() + queueLockedState() + mocks.enqueue.mockRejectedValueOnce(new Error('Database unavailable')) + await expect(disableInbox('workspace-1')).rejects.toThrow('Database unavailable') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mocks.process).not.toHaveBeenCalled() + }) + + it('retains configuration when provider identity cannot be read', async () => { + queueState() + mocks.getInbox.mockRejectedValueOnce(new Error('Service unavailable')) + await expect(disableInbox('workspace-1')).rejects.toThrow('Service unavailable') + expect(dbChainMockFns.transaction).not.toHaveBeenCalled() + }) + + it('disables processing while retaining the exact resource identities for retry', async () => { + queueState() + queueLockedState() + await disableInbox('workspace-1') + expect(mocks.enqueue).toHaveBeenCalledWith(expect.anything(), { + inboxId: oldState.providerId, + inboxCreatedAt: createdAt, + webhookId: 'old-hook', + }) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ inboxEnabled: false, inboxProviderId: null }) + ) + expect(mocks.process).toHaveBeenCalledWith('cleanup-event') + }) + + it('cleans up an already-absent inbox webhook', async () => { + queueState() + queueLockedState() + mocks.getInbox.mockResolvedValueOnce(null) + await disableInbox('workspace-1') + expect(mocks.enqueue).toHaveBeenCalledWith(expect.anything(), { + inboxId: null, + inboxCreatedAt: null, + webhookId: 'old-hook', + }) + }) + + it('refuses duplicate enable before provisioning', async () => { + queueState() + await expect(enableInbox('workspace-1')).rejects.toThrow('already configured') + expect(mocks.createInbox).not.toHaveBeenCalled() + }) + + it('does not let a losing concurrent enable delete the winning configuration', async () => { + queueTableRows(schemaMock.workspace, [emptyState]) + queueLockedState() + await expect(enableInbox('workspace-1')).rejects.toThrow('Inbox settings changed') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(mocks.enqueue).toHaveBeenCalledWith( + expect.anything(), + { + inboxId: newInbox.inbox_id, + inboxCreatedAt: createdAt, + webhookId: 'new-hook', + }, + expect.any(Date) + ) + }) + + it('rolls back new resources after a database commit failure', async () => { + queueTableRows(schemaMock.workspace, [emptyState]) + dbChainMockFns.transaction.mockRejectedValueOnce(new Error('Commit failed')) + await expect(enableInbox('workspace-1')).rejects.toThrow('Commit failed') + expect(mocks.enqueue).toHaveBeenCalledWith( + expect.anything(), + { + inboxId: newInbox.inbox_id, + inboxCreatedAt: createdAt, + webhookId: 'new-hook', + }, + expect.any(Date) + ) + expect(mocks.process).toHaveBeenCalledWith('cleanup-event', { expedite: true }) + }) + + it('directly rolls back uninstalled resources if rollback cannot be recorded', async () => { + queueTableRows(schemaMock.workspace, [emptyState]) + mocks.enqueue.mockRejectedValueOnce(new Error('Database unavailable')) + await expect(enableInbox('workspace-1')).rejects.toThrow('Database unavailable') + expect(dbChainMockFns.transaction).not.toHaveBeenCalled() + expect(mocks.deleteWebhook).toHaveBeenCalledWith('new-hook') + expect(mocks.deleteInbox).toHaveBeenCalledWith(newInbox.inbox_id) + }) + + it('refuses activation once a cleanup worker owns its resources', async () => { + queueTableRows(schemaMock.workspace, [emptyState]) + queueTableRows(schemaMock.workspace, [{ id: 'workspace-1' }]) + queueTableRows(schemaMock.workspace, [emptyState]) + mocks.cancel.mockRejectedValueOnce(new Error('Inbox setup expired')) + await expect(enableInbox('workspace-1')).rejects.toThrow('Inbox setup expired') + expect(dbChainMockFns.set).not.toHaveBeenCalled() + expect(mocks.deleteInbox).not.toHaveBeenCalled() + }) + + it('rolls back a created inbox when its response is missing the creation timestamp', async () => { + queueTableRows(schemaMock.workspace, [emptyState]) + mocks.createInbox.mockResolvedValueOnce({ inbox_id: newInbox.inbox_id }) + await expect(enableInbox('workspace-1')).rejects.toThrow( + 'Email service returned an invalid inbox' + ) + expect(mocks.deleteInbox).toHaveBeenCalledWith(newInbox.inbox_id) + expect(mocks.createWebhook).not.toHaveBeenCalled() + expect(dbChainMockFns.transaction).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/mothership/inbox/lifecycle.ts b/apps/sim/lib/mothership/inbox/lifecycle.ts index d92c916ec38..cb0f4a5867f 100644 --- a/apps/sim/lib/mothership/inbox/lifecycle.ts +++ b/apps/sim/lib/mothership/inbox/lifecycle.ts @@ -1,131 +1,198 @@ -import { db, mothershipInboxWebhook, workspace } from '@sim/db' +import { db } from '@sim/db' +import { mothershipInboxWebhook, workspace } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' import { eq } from 'drizzle-orm' import { getBaseUrl } from '@/lib/core/utils/urls' import * as agentmail from '@/lib/mothership/inbox/agentmail-client' -import type { InboxConfig } from '@/lib/mothership/inbox/types' +import { + cancelInboxCleanup, + enqueueInboxCleanup, + type InboxCleanupPayload, + processInboxCleanupNow, +} from '@/lib/mothership/inbox/cleanup-outbox' +import type { AgentMailInbox, AgentMailWebhook, InboxConfig } from '@/lib/mothership/inbox/types' const logger = createLogger('InboxLifecycle') +type InboxExecutor = Pick -/** - * Enable inbox for a workspace: - * 1. Create AgentMail inbox (with optional custom username) - * 2. Create AgentMail webhook scoped to this inbox - * 3. Store inbox details + webhook secret in DB - * 4. Update workspace.inboxEnabled = true - */ -export async function enableInbox( - workspaceId: string, - opts?: { username?: string } -): Promise { - const inbox = await agentmail.createInbox({ - username: opts?.username, - displayName: 'Sim', - }) +async function loadInbox(executor: InboxExecutor, workspaceId: string, lock = false) { + if (lock) { + await executor + .select({ id: workspace.id }) + .from(workspace) + .where(eq(workspace.id, workspaceId)) + .limit(1) + .for('update') + } + const query = executor + .select({ + enabled: workspace.inboxEnabled, + address: workspace.inboxAddress, + providerId: workspace.inboxProviderId, + webhookId: mothershipInboxWebhook.webhookId, + }) + .from(workspace) + .leftJoin(mothershipInboxWebhook, eq(mothershipInboxWebhook.workspaceId, workspace.id)) + .where(eq(workspace.id, workspaceId)) + .limit(1) + const [current] = await query + if (!current) throw new Error('Workspace not found') + return current +} - logger.info('AgentMail createInbox response', { inbox: JSON.stringify(inbox) }) +type InboxState = Awaited> - if (!inbox?.inbox_id) { - throw new Error('AgentMail createInbox response missing inbox_id') +function assertUnchanged(current: InboxState, expected: InboxState): void { + if ( + current.enabled !== expected.enabled || + current.providerId !== expected.providerId || + current.webhookId !== expected.webhookId + ) { + throw new Error('Inbox settings changed. Refresh and try again.') + } +} + +async function captureCleanup(current: InboxState): Promise { + const inbox = current.providerId ? await agentmail.getInbox(current.providerId) : null + return { + inboxId: inbox?.inbox_id ?? null, + inboxCreatedAt: inbox?.created_at ?? null, + webhookId: current.webhookId, + } +} + +/** Only used before activation can have committed, so direct rollback cannot delete a live inbox. */ +async function rollbackUninstalledResources( + inbox: AgentMailInbox, + webhook: AgentMailWebhook | null +) { + const results = await Promise.allSettled([ + ...(webhook ? [agentmail.deleteWebhook(webhook.webhook_id)] : []), + agentmail.deleteInbox(inbox.inbox_id), + ]) + if (results.some((result) => result.status === 'rejected' || !result.value)) { + logger.error('Inbox provisioning rollback needs reconciliation', { + inboxId: inbox.inbox_id, + webhookId: webhook?.webhook_id, + }) } +} + +async function rollbackProvisioning(inbox: AgentMailInbox, webhook: AgentMailWebhook | null) { + try { + const eventId = await enqueueInboxCleanup(db, { + inboxId: inbox.inbox_id, + inboxCreatedAt: inbox.created_at, + webhookId: webhook?.webhook_id ?? null, + }) + await processInboxCleanupNow(eventId) + } catch (error) { + await rollbackUninstalledResources(inbox, webhook) + logger.error('Failed to queue inbox provisioning rollback', { + inboxId: inbox.inbox_id, + webhookId: webhook?.webhook_id, + error, + }) + } +} - let webhook: Awaited> | null = null +async function provisionInbox(username?: string) { + const inbox = await agentmail.createInbox({ username, displayName: 'Sim' }) + if (!inbox?.inbox_id) { + throw new Error('Email service returned an invalid inbox') + } + if (!inbox.created_at) { + await rollbackUninstalledResources(inbox, null) + throw new Error('Email service returned an invalid inbox') + } try { - /** - * The receiver routes a delivery to this workspace by the `message.inbox_id` - * in the envelope, so it can only accept event types that carry a message. - * Adding one that does not would make those deliveries unroutable. - */ - webhook = await agentmail.createWebhook({ + const webhook = await agentmail.createWebhook({ url: `${getBaseUrl()}/api/webhooks/agentmail`, eventTypes: ['message.received'], inboxIds: [inbox.inbox_id], }) - - await db.insert(mothershipInboxWebhook).values({ - id: generateId(), - workspaceId, - webhookId: webhook.webhook_id, - secret: webhook.secret, - }) - - await db - .update(workspace) - .set({ - inboxEnabled: true, - inboxAddress: inbox.inbox_id, - inboxProviderId: inbox.inbox_id, - updatedAt: new Date(), - }) - .where(eq(workspace.id, workspaceId)) - - logger.info('Inbox enabled', { workspaceId, address: inbox.inbox_id }) - - return { - enabled: true, - address: inbox.inbox_id, - providerId: inbox.inbox_id, - } + return { inbox, webhook } } catch (error) { - try { - if (webhook) await agentmail.deleteWebhook(webhook.webhook_id) - await agentmail.deleteInbox(inbox.inbox_id) - await db - .delete(mothershipInboxWebhook) - .where(eq(mothershipInboxWebhook.workspaceId, workspaceId)) - } catch (rollbackError) { - logger.error('Failed to rollback AgentMail resources', { rollbackError }) - } + await rollbackProvisioning(inbox, null) throw error } } -/** - * Disable inbox: - * 1. Delete AgentMail webhook - * 2. Delete AgentMail inbox - * 3. Clear workspace inbox columns - * 4. Delete mothershipInboxWebhook row - */ -export async function disableInbox(workspaceId: string): Promise { - const [[ws], [webhookRow]] = await Promise.all([ - db - .select({ inboxProviderId: workspace.inboxProviderId }) - .from(workspace) - .where(eq(workspace.id, workspaceId)) - .limit(1), - db - .select({ webhookId: mothershipInboxWebhook.webhookId }) - .from(mothershipInboxWebhook) - .where(eq(mothershipInboxWebhook.workspaceId, workspaceId)) - .limit(1), - ]) - - const deletePromises: Promise[] = [] - if (webhookRow) { - deletePromises.push( - agentmail.deleteWebhook(webhookRow.webhookId).catch((error) => { - logger.warn('Failed to delete AgentMail webhook', { error }) - }) +async function installInbox( + workspaceId: string, + expected: InboxState, + username?: string, + cleanup?: InboxCleanupPayload +): Promise { + const { inbox, webhook } = await provisionInbox(username) + let rollbackEventId: string + try { + rollbackEventId = await enqueueInboxCleanup( + db, + { + inboxId: inbox.inbox_id, + inboxCreatedAt: inbox.created_at, + webhookId: webhook.webhook_id, + }, + new Date(Date.now() + 5 * 60_000) ) + } catch (error) { + await rollbackUninstalledResources(inbox, webhook) + throw error } - if (ws?.inboxProviderId) { - deletePromises.push( - agentmail.deleteInbox(ws.inboxProviderId).catch((error) => { - logger.warn('Failed to delete AgentMail inbox', { error }) + let cleanupEventId: string | undefined + try { + await db.transaction(async (tx) => { + assertUnchanged(await loadInbox(tx, workspaceId, true), expected) + await cancelInboxCleanup(tx, rollbackEventId) + if (cleanup) cleanupEventId = await enqueueInboxCleanup(tx, cleanup) + await tx + .delete(mothershipInboxWebhook) + .where(eq(mothershipInboxWebhook.workspaceId, workspaceId)) + await tx.insert(mothershipInboxWebhook).values({ + id: generateId(), + workspaceId, + webhookId: webhook.webhook_id, + secret: webhook.secret, }) - ) + await tx + .update(workspace) + .set({ + inboxEnabled: true, + inboxAddress: inbox.inbox_id, + inboxProviderId: inbox.inbox_id, + updatedAt: new Date(), + }) + .where(eq(workspace.id, workspaceId)) + }) + } catch (error) { + await processInboxCleanupNow(rollbackEventId, { expedite: true }) + throw error } - await Promise.all(deletePromises) + if (cleanupEventId) await processInboxCleanupNow(cleanupEventId) + return { enabled: true, address: inbox.inbox_id, providerId: inbox.inbox_id } +} - /** - * Atomic so the two rows cannot disagree. `workspace.inboxProviderId` is - * uniquely indexed, so a half-applied disable would strand the id of an - * AgentMail inbox that no longer exists — and the next workspace to claim that - * same address would then fail to enable at all. - */ - await db.transaction(async (tx) => { +/** Provisions resources before atomically activating their inbox and webhook configuration. */ +export async function enableInbox( + workspaceId: string, + opts?: { username?: string } +): Promise { + const current = await loadInbox(db, workspaceId) + if (current.enabled || current.providerId || current.webhookId) { + throw new Error('Inbox is already configured. Refresh and try again.') + } + return installInbox(workspaceId, current, opts?.username) +} + +/** Stops mail processing atomically and retains provider cleanup in the outbox until it succeeds. */ +export async function disableInbox(workspaceId: string): Promise { + const current = await loadInbox(db, workspaceId) + const cleanup = await captureCleanup(current) + const eventId = await db.transaction(async (tx) => { + assertUnchanged(await loadInbox(tx, workspaceId, true), current) + const eventId = await enqueueInboxCleanup(tx, cleanup) await tx .delete(mothershipInboxWebhook) .where(eq(mothershipInboxWebhook.workspaceId, workspaceId)) @@ -138,20 +205,17 @@ export async function disableInbox(workspaceId: string): Promise { updatedAt: new Date(), }) .where(eq(workspace.id, workspaceId)) + return eventId }) - - logger.info('Inbox disabled', { workspaceId }) + await processInboxCleanupNow(eventId) } -/** - * Update inbox address (regenerate): - * 1. Disable old inbox - * 2. Enable new inbox with new username - */ +/** Keeps the existing inbox intact until its replacement is provisioned and committed. */ export async function updateInboxAddress( workspaceId: string, newUsername: string ): Promise { - await disableInbox(workspaceId) - return enableInbox(workspaceId, { username: newUsername }) + const current = await loadInbox(db, workspaceId) + const cleanup = await captureCleanup(current) + return installInbox(workspaceId, current, newUsername, cleanup) }