diff --git a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts index 3ff1d479cd0..6ac51d538f5 100644 --- a/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts +++ b/apps/sim/app/api/knowledge/[id]/connectors/[connectorId]/route.ts @@ -12,6 +12,10 @@ import { hasWorkspaceLiveSyncAccess } from '@/lib/billing/core/subscription' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { resolveCredentialTokenIdentity } from '@/lib/credentials/access' +import { + preserveServerOwnedSourceConfig, + sanitizeConnectorSourceConfig, +} from '@/lib/knowledge/connectors/source-config' import { deleteDocumentStorageFiles } from '@/lib/knowledge/documents/service' import { cleanupUnusedTagDefinitions } from '@/lib/knowledge/tags/service' import { captureServerEvent } from '@/lib/posthog/server' @@ -103,6 +107,15 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Rout if (!parsed.success) return parsed.response const body = parsed.data.body + /** + * Guarded on throughout instead of `body.sourceConfig` so the sanitized value is + * the only one that can reach validation or the database. + */ + const sourceConfigUpdate = + body.sourceConfig === undefined ? undefined : sanitizeConnectorSourceConfig(body.sourceConfig) + /** Sanitized update plus the server-owned keys carried over from the stored row. */ + let sourceConfigToPersist: Record | undefined + if ( body.syncIntervalMinutes !== undefined && body.syncIntervalMinutes > 0 && @@ -124,7 +137,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Rout } } - if (body.sourceConfig !== undefined) { + if (sourceConfigUpdate !== undefined) { const existingRows = await db .select() .from(knowledgeConnector) @@ -143,6 +156,10 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Rout } const existing = existingRows[0] + sourceConfigToPersist = preserveServerOwnedSourceConfig( + sourceConfigUpdate, + existing.sourceConfig + ) const connectorConfig = CONNECTOR_REGISTRY[existing.connectorType] if (!connectorConfig) { @@ -152,60 +169,84 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Rout ) } - let accessToken: string | null = null - if (connectorConfig.auth.mode === 'apiKey') { - if (!existing.encryptedApiKey) { - return NextResponse.json( - { error: 'API key not found. Please reconfigure the connector.' }, - { status: 400 } - ) - } - accessToken = (await decryptApiKey(existing.encryptedApiKey)).decrypted - } else { - if (!existing.credentialId) { - return NextResponse.json( - { error: 'OAuth credential not found. Please reconfigure the connector.' }, - { status: 400 } - ) + const connectorWorkspaceId = writeCheck.knowledgeBase.workspaceId + if (!connectorWorkspaceId) { + return NextResponse.json( + { error: 'Knowledge base is missing workspace context' }, + { status: 409 } + ) + } + + /** + * Empty for `sim` connectors, which have no credential. Deliberately not + * guarded with a falsy check afterwards: every failure mode below returns its + * own response, so a falsy token here would only ever be a valid `sim` one. + */ + let accessToken = '' + + switch (connectorConfig.auth.mode) { + case 'sim': + break + + case 'apiKey': { + if (!existing.encryptedApiKey) { + return NextResponse.json( + { error: 'API key not found. Please reconfigure the connector.' }, + { status: 400 } + ) + } + accessToken = (await decryptApiKey(existing.encryptedApiKey)).decrypted + break } - const connectorWorkspaceId = writeCheck.knowledgeBase.workspaceId - if (!connectorWorkspaceId) { - return NextResponse.json( - { error: 'Knowledge base is missing workspace context' }, - { status: 409 } + + case 'oauth': { + if (!existing.credentialId) { + return NextResponse.json( + { error: 'OAuth credential not found. Please reconfigure the connector.' }, + { status: 400 } + ) + } + /** + * Resolve the credential's own account owner, not the knowledge base owner: + * workspace credentials are shared, and token reads are scoped to + * `account.userId`. + */ + const identity = await resolveCredentialTokenIdentity( + existing.credentialId, + connectorWorkspaceId ) - } - /** - * Resolve the credential's own account owner, not the knowledge base owner: - * workspace credentials are shared, and token reads are scoped to - * `account.userId`. - */ - const identity = await resolveCredentialTokenIdentity( - existing.credentialId, - connectorWorkspaceId - ) - if (!identity) { - return NextResponse.json( - { error: 'Credential is no longer usable in this workspace. Please reconnect it.' }, - { status: 400 } + if (!identity) { + return NextResponse.json( + { error: 'Credential is no longer usable in this workspace. Please reconnect it.' }, + { status: 400 } + ) + } + const refreshed = await refreshAccessTokenIfNeeded( + existing.credentialId, + // Service accounts mint their own token and ignore the acting user. + identity.kind === 'oauth' ? identity.userId : auth.userId, + `patch-${connectorId}` ) + if (!refreshed) { + return NextResponse.json( + { error: 'Failed to refresh access token. Please reconnect your account.' }, + { status: 401 } + ) + } + accessToken = refreshed + break } - accessToken = await refreshAccessTokenIfNeeded( - existing.credentialId, - // Service accounts mint their own token and ignore the acting user. - identity.kind === 'oauth' ? identity.userId : auth.userId, - `patch-${connectorId}` - ) - } - if (!accessToken) { - return NextResponse.json( - { error: 'Failed to refresh access token. Please reconnect your account.' }, - { status: 401 } - ) + default: { + const _exhaustive: never = connectorConfig.auth + return NextResponse.json({ error: 'Unsupported connector auth mode' }, { status: 400 }) + } } - const validation = await connectorConfig.validateConfig(accessToken, body.sourceConfig) + const validation = await connectorConfig.validateConfig(accessToken, sourceConfigUpdate, { + workspaceId: connectorWorkspaceId, + knowledgeBaseId, + }) if (!validation.valid) { return NextResponse.json( { error: validation.error || 'Invalid source configuration' }, @@ -215,8 +256,8 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Rout } const updates: Record = { updatedAt: new Date() } - if (body.sourceConfig !== undefined) { - updates.sourceConfig = body.sourceConfig + if (sourceConfigToPersist !== undefined) { + updates.sourceConfig = sourceConfigToPersist } if (body.syncIntervalMinutes !== undefined) { updates.syncIntervalMinutes = body.syncIntervalMinutes diff --git a/apps/sim/app/api/knowledge/[id]/connectors/route.ts b/apps/sim/app/api/knowledge/[id]/connectors/route.ts index b7df3198990..a95642703f6 100644 --- a/apps/sim/app/api/knowledge/[id]/connectors/route.ts +++ b/apps/sim/app/api/knowledge/[id]/connectors/route.ts @@ -17,6 +17,7 @@ import { hasWorkspaceLiveSyncAccess } from '@/lib/billing/core/subscription' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { dispatchSync } from '@/lib/knowledge/connectors/queue' +import { sanitizeConnectorSourceConfig } from '@/lib/knowledge/connectors/source-config' import { allocateTagSlots } from '@/lib/knowledge/constants' import { createTagDefinition } from '@/lib/knowledge/tags/service' import { captureServerEvent } from '@/lib/posthog/server' @@ -140,35 +141,65 @@ export const POST = withRouteHandler( let resolvedCredentialId: string | null = null let resolvedEncryptedApiKey: string | null = null - let accessToken: string + let accessToken = '' + + switch (connectorConfig.auth.mode) { + case 'sim': + /** + * Sim connectors read this workspace's own data, so there is nothing to + * store. Reject a supplied credential rather than ignoring it: persisting + * an unused `credentialId` would leave a connector row pointing at another + * workspace's credential, and an unused encrypted key is dead secret material. + */ + if (credentialId || apiKey) { + return NextResponse.json( + { error: 'This source reads your workspace directly and takes no credential' }, + { status: 400 } + ) + } + break - if (connectorConfig.auth.mode === 'apiKey') { - if (!apiKey) { - return NextResponse.json({ error: 'API key is required' }, { status: 400 }) - } - accessToken = apiKey - } else { - if (!credentialId) { - return NextResponse.json({ error: 'Credential is required' }, { status: 400 }) - } + case 'apiKey': + if (!apiKey) { + return NextResponse.json({ error: 'API key is required' }, { status: 400 }) + } + accessToken = apiKey + break - const credential = await getCredential(requestId, credentialId, auth.userId) - if (!credential) { - return NextResponse.json({ error: 'Credential not found' }, { status: 400 }) - } + case 'oauth': { + if (!credentialId) { + return NextResponse.json({ error: 'Credential is required' }, { status: 400 }) + } - if (!credential.accessToken) { - return NextResponse.json( - { error: 'Credential has no access token. Please reconnect your account.' }, - { status: 400 } - ) + const credential = await getCredential(requestId, credentialId, auth.userId) + if (!credential) { + return NextResponse.json({ error: 'Credential not found' }, { status: 400 }) + } + + if (!credential.accessToken) { + return NextResponse.json( + { error: 'Credential has no access token. Please reconnect your account.' }, + { status: 400 } + ) + } + + accessToken = credential.accessToken + resolvedCredentialId = credentialId + break } - accessToken = credential.accessToken - resolvedCredentialId = credentialId + default: { + const _exhaustive: never = connectorConfig.auth + return NextResponse.json({ error: 'Unsupported connector auth mode' }, { status: 400 }) + } } - const configValidation = await connectorConfig.validateConfig(accessToken, sourceConfig) + const safeSourceConfig = sanitizeConnectorSourceConfig(sourceConfig) + + const configValidation = await connectorConfig.validateConfig(accessToken, safeSourceConfig, { + workspaceId: kbWorkspaceId, + knowledgeBaseId, + }) if (!configValidation.valid) { return NextResponse.json( { error: configValidation.error || 'Invalid source configuration' }, @@ -176,7 +207,7 @@ export const POST = withRouteHandler( ) } - let finalSourceConfig: Record = { ...sourceConfig } + let finalSourceConfig: Record = { ...safeSourceConfig } if (connectorConfig.auth.mode === 'apiKey' && apiKey) { const { encrypted } = await encryptApiKey(apiKey) @@ -187,7 +218,7 @@ export const POST = withRouteHandler( let newTagSlots: Record = {} if (connectorConfig.tagDefinitions?.length) { - const disabledIds = new Set((sourceConfig.disabledTagIds as string[] | undefined) ?? []) + const disabledIds = new Set((safeSourceConfig.disabledTagIds as string[] | undefined) ?? []) const enabledDefs = connectorConfig.tagDefinitions.filter((td) => !disabledIds.has(td.id)) const existingDefs = await db diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/markdown-parse.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/markdown-parse.test.ts index 0067ef31f47..68ef535e997 100644 --- a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/markdown-parse.test.ts +++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/markdown-parse.test.ts @@ -275,6 +275,10 @@ describe('chunked parse — property test over randomized documents', () => { } } expect(failures).toEqual([]) - // 400 docs each parsed+serialized twice — generous timeout so it can't flake under parallel load. - }, 30000) + /** + * 400 docs each parsed+serialized twice. ~10s in isolation, so 30s left barely 3x + * headroom and still timed out during a full-suite run; 60s restores real margin + * without dropping seeds, since coverage here is the number of documents fuzzed. + */ + }, 60000) }) diff --git a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/add-connector-modal/add-connector-modal.tsx b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/add-connector-modal/add-connector-modal.tsx index 705746d348f..6f3cafa9094 100644 --- a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/add-connector-modal/add-connector-modal.tsx +++ b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/add-connector-modal/add-connector-modal.tsx @@ -38,7 +38,11 @@ import { useWorkspaceHostContext } from '@/app/workspace/[workspaceId]/providers import { getBlock } from '@/blocks' import { getTileIconColorClass } from '@/blocks/icon-color' import { CONNECTOR_META_REGISTRY } from '@/connectors/registry' -import type { ConnectorMeta } from '@/connectors/types' +import { + type ConnectorAuthConfig, + type ConnectorMeta, + collectsCredential, +} from '@/connectors/types' import { useCreateConnector } from '@/hooks/queries/kb/connectors' import { useOAuthCredentials } from '@/hooks/queries/oauth/oauth-credentials' import { useCredentialRefreshTriggers } from '@/hooks/use-credential-refresh-triggers' @@ -81,7 +85,8 @@ export function AddConnectorModal({ const hasMaxAccess = hasWorkspaceMaxConnectorAccess(ownerBilling) const connectorConfig = selectedType ? CONNECTOR_META_REGISTRY[selectedType] : null - const isApiKeyMode = connectorConfig?.auth.mode === 'apiKey' + const authMode = connectorConfig?.auth.mode + const isApiKeyMode = authMode === 'apiKey' const connectorProviderId = useMemo( () => connectorConfig && connectorConfig.auth.mode === 'oauth' @@ -95,7 +100,8 @@ export function AddConnectorModal({ isLoading: credentialsLoading, refetch: refetchCredentials, } = useOAuthCredentials(connectorProviderId ?? undefined, { - enabled: Boolean(connectorConfig) && !isApiKeyMode, + // Non-null only for `oauth`, so this covers `apiKey` and `sim` in one condition. + enabled: Boolean(connectorProviderId), workspaceId, }) @@ -159,10 +165,8 @@ export function AddConnectorModal({ const canSubmit = useMemo(() => { if (!connectorConfig) return false - if (isApiKeyMode) { - if (!apiKeyValue.trim()) return false - } else { - if (!effectiveCredentialId) return false + if (collectsCredential(connectorConfig.auth)) { + if (isApiKeyMode ? !apiKeyValue.trim() : !effectiveCredentialId) return false } for (const field of connectorConfig.configFields) { @@ -207,7 +211,11 @@ export function AddConnectorModal({ { knowledgeBaseId, connectorType: selectedType, - ...(isApiKeyMode ? { apiKey: apiKeyValue } : { credentialId: effectiveCredentialId! }), + ...(authMode === 'sim' + ? {} + : isApiKeyMode + ? { apiKey: apiKeyValue } + : { credentialId: effectiveCredentialId! }), sourceConfig: finalSourceConfig, syncIntervalMinutes: syncInterval, }, @@ -231,6 +239,71 @@ export function AddConnectorModal({ ) }, [searchTerm]) + /** + * Exhaustive over `ConnectorAuthConfig` so a new auth mode fails to compile here + * rather than silently falling through to the OAuth account picker. + */ + const renderAuthField = (auth: ConnectorAuthConfig) => { + switch (auth.mode) { + case 'sim': + /** Nothing to collect: the connector reads this workspace directly. */ + return null + + case 'apiKey': + return ( + + setApiKeyValue(e.target.value)} + onFocus={() => setApiKeyFocused(true)} + onBlur={() => setApiKeyFocused(false)} + placeholder={auth.placeholder || 'Enter API key'} + /> + + ) + + case 'oauth': + return ( + + ({ + label: cred.name || cred.provider, + value: cred.id, + icon: connectorConfig?.icon, + }) + ), + { + label: + credentials.length > 0 + ? `Connect another ${connectorConfig?.name} account` + : `Connect ${connectorConfig?.name} account`, + value: '__connect_new__', + icon: Plus, + onSelect: () => setShowOAuthModal(true), + }, + ]} + value={effectiveCredentialId ?? undefined} + onChange={(value) => setSelectedCredentialId(value)} + onOpenChange={(isOpen) => { + if (isOpen) void refetchCredentials() + }} + placeholder={`Select ${connectorConfig?.name} account`} + isLoading={credentialsLoading} + /> + + ) + + default: { + const _exhaustive: never = auth + return null + } + } + } + return ( <> ) : connectorConfig ? ( <> - {isApiKeyMode ? ( - - setApiKeyValue(e.target.value)} - onFocus={() => setApiKeyFocused(true)} - onBlur={() => setApiKeyFocused(false)} - placeholder={ - connectorConfig.auth.mode === 'apiKey' && connectorConfig.auth.placeholder - ? connectorConfig.auth.placeholder - : 'Enter API key' - } - /> - - ) : ( - - ({ - label: cred.name || cred.provider, - value: cred.id, - icon: connectorConfig.icon, - }) - ), - { - label: - credentials.length > 0 - ? `Connect another ${connectorConfig.name} account` - : `Connect ${connectorConfig.name} account`, - value: '__connect_new__', - icon: Plus, - onSelect: () => setShowOAuthModal(true), - }, - ]} - value={effectiveCredentialId ?? undefined} - onChange={(value) => setSelectedCredentialId(value)} - onOpenChange={(isOpen) => { - if (isOpen) void refetchCredentials() - }} - placeholder={`Select ${connectorConfig.name} account`} - isLoading={credentialsLoading} - /> - - )} + {renderAuthField(connectorConfig.auth)} onFieldChange(field.id, value)} credentialId={credentialId} + requiresCredential={collectsCredential(connectorConfig.auth)} sourceConfig={sourceConfig} configFields={connectorConfig.configFields} canonicalModes={canonicalModes} diff --git a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connector-selector-field/connector-selector-field.tsx b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connector-selector-field/connector-selector-field.tsx index e9053558edf..a67354e624c 100644 --- a/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connector-selector-field/connector-selector-field.tsx +++ b/apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connector-selector-field/connector-selector-field.tsx @@ -2,6 +2,7 @@ import { useMemo, useState } from 'react' import { ChipCombobox, type ComboboxOption, Loader } from '@sim/emcn' +import { useParams } from 'next/navigation' import { SEARCH_DEBOUNCE_MS } from '@/lib/url-state' import { SELECTOR_CONTEXT_FIELDS } from '@/lib/workflows/subblocks/context' import type { @@ -24,6 +25,11 @@ interface ConnectorSelectorFieldProps { value: ConfigFieldValue onChange: (value: ConfigFieldValue) => void credentialId: string | null + /** + * False for `sim` connectors, which read this workspace directly and have no + * account to connect. Their selectors are ready as soon as their deps resolve. + */ + requiresCredential?: boolean sourceConfig: ConfigFieldMap configFields: ConnectorConfigField[] canonicalModes: Record @@ -35,16 +41,24 @@ export function ConnectorSelectorField({ value, onChange, credentialId, + requiresCredential = true, sourceConfig, configFields, canonicalModes, disabled, }: ConnectorSelectorFieldProps) { + const { workspaceId } = useParams<{ workspaceId: string }>() const isMulti = Boolean(field.multi) const [searchTerm, setSearchTerm] = useState('') const context = useMemo(() => { const ctx: SelectorContext = {} + /** + * Set before the dependsOn loop, which can only write keys listed in + * SELECTOR_CONTEXT_FIELDS — `workspaceId` is not one, so a config field can + * never overwrite it. Workspace-scoped selectors (`sim.*`) are inert without it. + */ + if (workspaceId) ctx.workspaceId = workspaceId if (credentialId) ctx.oauthCredential = credentialId if (field.mimeType) ctx.mimeType = field.mimeType @@ -59,7 +73,15 @@ export function ConnectorSelectorField({ } return ctx - }, [credentialId, field.mimeType, field.dependsOn, sourceConfig, configFields, canonicalModes]) + }, [ + workspaceId, + credentialId, + field.mimeType, + field.dependsOn, + sourceConfig, + configFields, + canonicalModes, + ]) const depsResolved = useMemo(() => { if (!field.dependsOn) return true @@ -69,7 +91,8 @@ export function ConnectorSelectorField({ ) }, [field.dependsOn, sourceConfig, configFields, canonicalModes]) - const isEnabled = !disabled && !!credentialId && depsResolved + const credentialSatisfied = !requiresCredential || !!credentialId + const isEnabled = !disabled && credentialSatisfied && depsResolved const { data: options = [], isLoading, @@ -154,13 +177,13 @@ export function ConnectorSelectorField({ onSearchChange={setSearchTerm} searchPlaceholder={`Search ${field.title.toLowerCase()}...`} placeholder={ - !credentialId + !credentialSatisfied ? 'Connect an account first' : !depsResolved ? `Select ${getDependencyLabel(field, configFields)} first` : field.placeholder || `Select ${field.title.toLowerCase()}` } - disabled={disabled || !credentialId || !depsResolved} + disabled={disabled || !credentialSatisfied || !depsResolved} emptyMessage={emptyMessage} /> ) @@ -175,13 +198,13 @@ export function ConnectorSelectorField({ onSearchChange={setSearchTerm} searchPlaceholder={`Search ${field.title.toLowerCase()}...`} placeholder={ - !credentialId + !credentialSatisfied ? 'Connect an account first' : !depsResolved ? `Select ${getDependencyLabel(field, configFields)} first` : field.placeholder || `Select ${field.title.toLowerCase()}` } - disabled={disabled || !credentialId || !depsResolved} + disabled={disabled || !credentialSatisfied || !depsResolved} emptyMessage={emptyMessage} /> ) diff --git a/apps/sim/components/icons.tsx b/apps/sim/components/icons.tsx index 606e5b0faf6..439ebfd13ab 100644 --- a/apps/sim/components/icons.tsx +++ b/apps/sim/components/icons.tsx @@ -8944,3 +8944,35 @@ export function ZohoDeskIcon(props: SVGProps) { ) } + +/** + * The "sim" logotype, sized for an icon slot. + * + * Same paths as `SimWordmark` in the landing navbar (v1.0 brand guide's + * `simLogotype--dark.svg`), re-exposed as a prop-taking icon so surfaces that + * size their icon via `className` - connector tiles, pickers - can use the real + * brand mark instead of a stand-in glyph. + * + * Filled with `var(--text-body)` rather than `currentColor` on purpose: this is a + * brand mark, so it stays solid ink (black on light, white on dark) and does not + * take on a caller's muted icon color. The wordmark is ~2:1, so it letterboxes + * inside a square slot rather than stretching. + */ +export function SimLogoIcon(props: SVGProps) { + return ( + + ) +} diff --git a/apps/sim/connectors/registry.server.ts b/apps/sim/connectors/registry.server.ts index ba870e2af41..f297f6d982b 100644 --- a/apps/sim/connectors/registry.server.ts +++ b/apps/sim/connectors/registry.server.ts @@ -42,6 +42,8 @@ import { salesforceConnector } from '@/connectors/salesforce' import { sentryConnector } from '@/connectors/sentry' import { servicenowConnector } from '@/connectors/servicenow' import { sharepointConnector } from '@/connectors/sharepoint' +import { simConversationsConnector } from '@/connectors/sim-conversations' +import { simFilesConnector } from '@/connectors/sim-files' import { slackConnector } from '@/connectors/slack' import { typeformConnector } from '@/connectors/typeform' import type { ConnectorRegistry } from '@/connectors/types' @@ -104,6 +106,8 @@ export const CONNECTOR_REGISTRY: ConnectorRegistry = { sentry: sentryConnector, servicenow: servicenowConnector, sharepoint: sharepointConnector, + sim_conversations: simConversationsConnector, + sim_files: simFilesConnector, slack: slackConnector, typeform: typeformConnector, webflow: webflowConnector, diff --git a/apps/sim/connectors/registry.test.ts b/apps/sim/connectors/registry.test.ts new file mode 100644 index 00000000000..5baa20cddf1 --- /dev/null +++ b/apps/sim/connectors/registry.test.ts @@ -0,0 +1,214 @@ +/** + * @vitest-environment node + * + * Structural invariants across every connector. These are the failure modes that + * compile cleanly and only surface at runtime — a connector missing from one of the + * two registries, a selector field whose manual twin drifted, an auth mode the UI + * cannot render. + */ +import { describe, expect, it } from 'vitest' +import { getSlotsForFieldType } from '@/lib/knowledge/constants' +import { CONNECTOR_META_REGISTRY } from '@/connectors/registry' +import { CONNECTOR_REGISTRY } from '@/connectors/registry.server' +import { type ConnectorConfigField, collectsCredential } from '@/connectors/types' +import { selectorRegistry } from '@/hooks/selectors/registry' + +const metaEntries = Object.entries(CONNECTOR_META_REGISTRY) +const runtimeEntries = Object.entries(CONNECTOR_REGISTRY) + +/** + * Pre-existing connectors that declare more tags of a type than there are slots, so + * `allocateTagSlots` always drops the ones it cannot place on a fresh knowledge base — + * silently, with only a server-side warning. + * + * - `azure_devops`: 8 text tags against 7 text slots (`path` / "File Path" is lost). + * - `google_calendar`: 3 date tags against 2 date slots. + * + * Every tag involved is genuinely populated by its connector's `mapTags`, so choosing + * which to cut is a product decision about that service's semantics rather than a + * mechanical fix, and is left to each connector's owner. Ratcheted rather than + * relaxed so the invariant still holds for every other connector. + */ +const OVERSUBSCRIBED_TAG_CONNECTORS = new Set(['azure_devops', 'google_calendar']) + +describe('connector registries', () => { + /** A connector present in only one registry breaks either the picker or the sync. */ + it('registers every connector in both the client and server registries', () => { + expect(Object.keys(CONNECTOR_REGISTRY).sort()).toEqual( + Object.keys(CONNECTOR_META_REGISTRY).sort() + ) + }) + + it('keys every connector by its own id', () => { + for (const [key, meta] of metaEntries) { + expect(meta.id, `${key} registry key must match meta.id`).toBe(key) + } + }) + + it('keeps both registries alphabetically ordered', () => { + const metaKeys = Object.keys(CONNECTOR_META_REGISTRY) + const runtimeKeys = Object.keys(CONNECTOR_REGISTRY) + expect(metaKeys).toEqual([...metaKeys].sort()) + expect(runtimeKeys).toEqual([...runtimeKeys].sort()) + }) +}) + +describe('connector config fields', () => { + /** + * The add-connector modal persists a canonical pair under its `canonicalParamId`, + * choosing whichever member matches the active mode. A selector without its manual + * twin leaves the advanced toggle with nothing to switch to; mismatched `required` + * lets a field be mandatory in one mode and optional in the other. + */ + it('pairs every selector field with a manual twin of matching requiredness', () => { + for (const [key, meta] of metaEntries) { + const selectors = meta.configFields.filter( + (field): field is ConnectorConfigField => field.type === 'selector' + ) + + for (const selector of selectors) { + expect( + selector.canonicalParamId, + `${key}.${selector.id} needs a canonicalParamId` + ).toBeTruthy() + expect(selector.mode, `${key}.${selector.id} must declare a mode`).toBe('basic') + + const twins = meta.configFields.filter( + (field) => + field.canonicalParamId === selector.canonicalParamId && field.id !== selector.id + ) + + expect(twins, `${key}.${selector.id} must have exactly one manual twin`).toHaveLength(1) + expect(twins[0].mode, `${key}.${twins[0].id} must be the advanced twin`).toBe('advanced') + expect( + Boolean(twins[0].required), + `${key}.${twins[0].id} requiredness must match its selector` + ).toBe(Boolean(selector.required)) + expect(Boolean(twins[0].multi), `${key}.${twins[0].id} multi must match its selector`).toBe( + Boolean(selector.multi) + ) + } + } + }) + + /** + * `getSelectorDefinition` throws on an unknown key, and only when the field renders. + * Catches a `SelectorKey` union widened without registering the definition. + */ + it('references only selector keys that exist in the selector registry', () => { + for (const [key, meta] of metaEntries) { + for (const field of meta.configFields) { + if (field.type !== 'selector' || !field.selectorKey) continue + expect( + Object.keys(selectorRegistry), + `${key}.${field.id} references unknown selector ${field.selectorKey}` + ).toContain(field.selectorKey) + } + } + }) + + it('gives every dropdown field options to choose from', () => { + for (const [key, meta] of metaEntries) { + for (const field of meta.configFields) { + if (field.type !== 'dropdown') continue + expect(field.options?.length, `${key}.${field.id} dropdown needs options`).toBeGreaterThan( + 0 + ) + } + } + }) +}) + +describe('connector tag definitions', () => { + /** Slots are allocated from `mapTags` output, so tags without it are never written. */ + it('implements mapTags wherever tag definitions are declared', () => { + for (const [key, connector] of runtimeEntries) { + if (!connector.tagDefinitions?.length) continue + expect(typeof connector.mapTags, `${key} declares tags but has no mapTags`).toBe('function') + } + }) + + it('uses unique tag ids and display names per connector', () => { + for (const [key, meta] of metaEntries) { + const ids = (meta.tagDefinitions ?? []).map((tag) => tag.id) + const names = (meta.tagDefinitions ?? []).map((tag) => tag.displayName) + expect(new Set(ids).size, `${key} has duplicate tag ids`).toBe(ids.length) + expect(new Set(names).size, `${key} has duplicate tag display names`).toBe(names.length) + } + }) + + /** + * Slots per field type are finite (`TAG_SLOT_CONFIG`: 7 text, 5 number, 2 date, 3 + * boolean). A connector declaring more tags of a type than there are slots + * guarantees at least one is silently dropped — `allocateTagSlots` only logs a + * warning and 422s when it could place nothing at all. + * + * Fitting within the budget is not the same as being a good neighbor: a connector + * claiming every slot of a type starves any other connector on the same knowledge + * base. That is a judgment call per connector, so it is not asserted here. + */ + it('declares no more tags of a type than there are slots for it', () => { + for (const [key, meta] of metaEntries) { + if (OVERSUBSCRIBED_TAG_CONNECTORS.has(key)) continue + + const countsByType = new Map() + for (const tag of meta.tagDefinitions ?? []) { + countsByType.set(tag.fieldType, (countsByType.get(tag.fieldType) ?? 0) + 1) + } + + for (const [fieldType, count] of countsByType) { + const capacity = getSlotsForFieldType(fieldType).length + expect(capacity, `${key} uses unknown tag field type "${fieldType}"`).toBeGreaterThan(0) + expect( + count, + `${key} declares ${count} ${fieldType} tags but only ${capacity} slots exist` + ).toBeLessThanOrEqual(capacity) + } + } + }) + + /** The exemption list must shrink, never grow. */ + it('has no unnecessary entries in the oversubscribed allowlist', () => { + for (const key of OVERSUBSCRIBED_TAG_CONNECTORS) { + expect(CONNECTOR_META_REGISTRY[key], `${key} is allowlisted but not registered`).toBeDefined() + } + expect(OVERSUBSCRIBED_TAG_CONNECTORS.size).toBeLessThanOrEqual(2) + }) +}) + +describe('sim-mode connectors', () => { + const simConnectors = runtimeEntries.filter(([, connector]) => connector.auth.mode === 'sim') + + it('ships the expected native connectors', () => { + expect(simConnectors.map(([key]) => key).sort()).toEqual(['sim_conversations', 'sim_files']) + }) + + /** The modal renders no auth row for these, so they must not need one. */ + it('collects no credential', () => { + for (const [key, connector] of simConnectors) { + expect(collectsCredential(connector.auth), `${key} must not collect a credential`).toBe(false) + } + }) + + /** + * Validation runs before any credential exists, with an empty access token. Anything + * that reached for a token here would throw on connector creation. + */ + it('validates an empty config without a token', async () => { + for (const [key, connector] of simConnectors) { + await expect( + connector.validateConfig('', {}), + `${key} must validate without a credential` + ).resolves.toMatchObject({ valid: true }) + } + }) + + it('rejects malformed numeric config', async () => { + await expect( + CONNECTOR_REGISTRY.sim_files.validateConfig('', { maxFiles: 'lots' }) + ).resolves.toMatchObject({ valid: false }) + await expect( + CONNECTOR_REGISTRY.sim_conversations.validateConfig('', { minMessages: '-1' }) + ).resolves.toMatchObject({ valid: false }) + }) +}) diff --git a/apps/sim/connectors/registry.ts b/apps/sim/connectors/registry.ts index b1fd50e9736..df9d451cea6 100644 --- a/apps/sim/connectors/registry.ts +++ b/apps/sim/connectors/registry.ts @@ -42,6 +42,8 @@ import { salesforceConnectorMeta } from '@/connectors/salesforce/meta' import { sentryConnectorMeta } from '@/connectors/sentry/meta' import { servicenowConnectorMeta } from '@/connectors/servicenow/meta' import { sharepointConnectorMeta } from '@/connectors/sharepoint/meta' +import { simConversationsConnectorMeta } from '@/connectors/sim-conversations/meta' +import { simFilesConnectorMeta } from '@/connectors/sim-files/meta' import { slackConnectorMeta } from '@/connectors/slack/meta' import { typeformConnectorMeta } from '@/connectors/typeform/meta' import type { ConnectorMeta, ConnectorMetaRegistry } from '@/connectors/types' @@ -104,6 +106,8 @@ export const CONNECTOR_META_REGISTRY: ConnectorMetaRegistry = { sentry: sentryConnectorMeta, servicenow: servicenowConnectorMeta, sharepoint: sharepointConnectorMeta, + sim_conversations: simConversationsConnectorMeta, + sim_files: simFilesConnectorMeta, slack: slackConnectorMeta, typeform: typeformConnectorMeta, webflow: webflowConnectorMeta, diff --git a/apps/sim/connectors/sim-conversations/index.ts b/apps/sim/connectors/sim-conversations/index.ts new file mode 100644 index 00000000000..3c6eae17680 --- /dev/null +++ b/apps/sim/connectors/sim-conversations/index.ts @@ -0,0 +1 @@ +export { simConversationsConnector } from '@/connectors/sim-conversations/sim-conversations' diff --git a/apps/sim/connectors/sim-conversations/meta.ts b/apps/sim/connectors/sim-conversations/meta.ts new file mode 100644 index 00000000000..b1692175957 --- /dev/null +++ b/apps/sim/connectors/sim-conversations/meta.ts @@ -0,0 +1,59 @@ +import { SimLogoIcon } from '@/components/icons' +import type { ConnectorMeta } from '@/connectors/types' + +export const simConversationsConnectorMeta: ConnectorMeta = { + id: 'sim_conversations', + name: 'Agent Conversations', + description: 'Sync agent block conversation memory so you can search and analyze what users ask', + version: '1.0.0', + icon: SimLogoIcon, + + auth: { mode: 'sim' }, + + configFields: [ + { + id: 'conversationIdPrefix', + title: 'Conversation ID Prefix', + type: 'short-input', + required: false, + placeholder: 'e.g. support-', + description: + 'Only conversations whose ID starts with this text are synced. Scoping is by ID rather than by workflow because one conversation ID can be shared across several agent blocks and workflows. Leave empty to sync every conversation in this workspace.', + }, + { + id: 'minMessages', + title: 'Minimum Messages', + type: 'short-input', + required: false, + placeholder: 'e.g. 2', + description: 'Skips conversations shorter than this. Defaults to 1.', + }, + { + id: 'maxConversations', + title: 'Max Conversations', + type: 'short-input', + required: false, + placeholder: 'e.g. 1000', + description: 'Caps how many conversations are indexed. Leave empty for no limit.', + }, + ], + + /** + * Deliberately absent, for the same reasons as `sim_files`: listing is one indexed + * local query, and an incremental run would disable deletion reconciliation + * (`shouldReconcileDeletions`), so conversations deleted at the source would linger + * in the knowledge base until someone ran a manual full resync. + */ + supportsIncrementalSync: false, + + /** + * `startedAt` is deliberately omitted. `DocumentTags` exposes only two date slots, + * so claiming both would leave none for any other connector on the same knowledge + * base — and `allocateTagSlots` would silently skip whichever it could not place. + */ + tagDefinitions: [ + { id: 'conversationId', displayName: 'Conversation ID', fieldType: 'text' }, + { id: 'messageCount', displayName: 'Message Count', fieldType: 'number' }, + { id: 'lastActivity', displayName: 'Last Activity', fieldType: 'date' }, + ], +} diff --git a/apps/sim/connectors/sim-conversations/sim-conversations.test.ts b/apps/sim/connectors/sim-conversations/sim-conversations.test.ts new file mode 100644 index 00000000000..d202dc760c7 --- /dev/null +++ b/apps/sim/connectors/sim-conversations/sim-conversations.test.ts @@ -0,0 +1,309 @@ +/** + * @vitest-environment node + */ +import { flattenMockConditions, type MockCondition } from '@sim/testing' +import { describe, expect, it } from 'vitest' +import { + buildConversationListingFilters, + type ConversationRow, + conversationToStub, + decodeCursor, + encodeCursor, + escapeLikePrefix, + parseOptionalPositiveInt, + renderTranscript, +} from '@/connectors/sim-conversations/sim-conversations' + +/** Shape the drizzle `sql` mock produces (see packages/testing database.mock). */ +interface SqlFragment { + strings?: readonly string[] + values?: unknown[] +} + +const BASE_ROW: ConversationRow = { + id: 'mem-1', + key: 'support-123', + createdAt: new Date('2026-01-01T00:00:00.000Z'), + updatedAt: new Date('2026-01-02T00:00:00.000Z'), + messageCount: 6, + approxBytes: 1024, + contentDigest: 'd41d8cd98f00b204e9800998ecf8427e', +} + +const META = { + conversationId: 'support-123', + startedAt: '2026-01-01T00:00:00.000Z', + lastActivity: '2026-01-02T00:00:00.000Z', +} + +function conditionsOf( + args: Parameters[0] +): MockCondition[] { + return buildConversationListingFilters(args).flatMap(flattenMockConditions) +} + +describe('escapeLikePrefix', () => { + /** + * The whole point of the filter. Unescaped, a prefix of `%` matches every + * conversation in the workspace, turning a narrow scope into a full export. + */ + it('escapes LIKE wildcards so they match literally', () => { + expect(escapeLikePrefix('%')).toBe('\\%') + expect(escapeLikePrefix('_')).toBe('\\_') + expect(escapeLikePrefix('100%_done')).toBe('100\\%\\_done') + }) + + it('escapes the escape character itself', () => { + expect(escapeLikePrefix('a\\b')).toBe('a\\\\b') + }) + + it('leaves ordinary prefixes untouched', () => { + expect(escapeLikePrefix('support-')).toBe('support-') + expect(escapeLikePrefix('')).toBe('') + }) +}) + +describe('buildConversationListingFilters', () => { + /** + * The builder takes no `sourceConfig`, so this proves only that the supplied + * workspace is bound. That the connector never READS `sourceConfig.workspaceId` + * is proven end to end by the sync harness, not here. + */ + it('binds the supplied workspace', () => { + const nodes = conditionsOf({ workspaceId: 'ws-real', prefix: '' }) + const workspaceClause = nodes.find((node) => node.left === 'workspaceId') + + expect(workspaceClause).toMatchObject({ type: 'eq', right: 'ws-real' }) + }) + + /** + * `Memory.fetchMemory` omits this filter; copying that omission would resurrect + * conversations a workspace has already deleted. + */ + it('excludes soft-deleted conversations', () => { + const nodes = conditionsOf({ workspaceId: 'ws-1', prefix: '' }) + expect(nodes.map((node) => node.column ?? node.left)).toContain('deletedAt') + }) + + it('adds a prefix filter only when a prefix is configured', () => { + expect(conditionsOf({ workspaceId: 'ws-1', prefix: '' })).toHaveLength(2) + expect(conditionsOf({ workspaceId: 'ws-1', prefix: 'support-' }).length).toBeGreaterThan(2) + }) + + /** + * Asserts the WIRING, not just that `escapeLikePrefix` works in isolation. + * Without this, deleting the escape call from the builder still passes every + * other test while a prefix of `%` exports every conversation in the workspace. + */ + it('binds the ESCAPED prefix as the LIKE parameter', () => { + const filters = buildConversationListingFilters({ workspaceId: 'ws-1', prefix: '100%_done' }) + const bound = JSON.stringify(filters.map((f) => (f as unknown as SqlFragment).values ?? null)) + + expect(bound).toContain('100\\\\%\\\\_done%') + expect(bound).not.toContain('"100%_done%"') + expect( + JSON.stringify(filters.map((f) => (f as unknown as SqlFragment).strings ?? null)) + ).toContain('ESCAPE') + }) + + /** The keyset direction must match ORDER BY — see the files connector's note. */ + it('flips the keyset comparison when the listing is descending', () => { + const cursor = { updatedAt: new Date('2026-01-01T00:00:00.000Z'), id: 'mem-1' } + const ascending = conditionsOf({ workspaceId: 'ws-1', prefix: '', cursor }) + const descending = conditionsOf({ workspaceId: 'ws-1', prefix: '', cursor, descending: true }) + + const branches = (nodes: MockCondition[]) => + nodes + .filter((n) => n.type === 'or') + .flatMap((n) => (n.conditions as MockCondition[]) ?? []) + .flatMap(flattenMockConditions) + + expect(branches(ascending).some((n) => n.type === 'gt')).toBe(true) + expect(branches(descending).some((n) => n.type === 'lt')).toBe(true) + expect(branches(descending).some((n) => n.type === 'gt')).toBe(false) + }) + + it('adds a keyset clause only when paginating', () => { + const first = conditionsOf({ workspaceId: 'ws-1', prefix: '' }) + const next = conditionsOf({ + workspaceId: 'ws-1', + prefix: '', + cursor: { updatedAt: new Date('2026-01-01T00:00:00.000Z'), id: 'mem-1' }, + }) + + expect(first.some((node) => node.type === 'or')).toBe(false) + expect(next.some((node) => node.type === 'or')).toBe(true) + }) +}) + +describe('renderTranscript', () => { + it('renders roles as headings in order', () => { + const content = renderTranscript(META, [ + { role: 'user', content: 'how do I reset my key?' }, + { role: 'assistant', content: 'Open Settings, then API Keys.' }, + ]) + + expect(content).toContain('# Conversation: support-123') + expect(content).toContain('## User\n\nhow do I reset my key?') + expect(content).toContain('## Assistant\n\nOpen Settings, then API Keys.') + expect(content.indexOf('## User')).toBeLessThan(content.indexOf('## Assistant')) + }) + + it('reports the true message count in the header', () => { + const content = renderTranscript(META, [ + { role: 'user', content: 'a' }, + { role: 'assistant', content: 'b' }, + { role: 'user', content: 'c' }, + ]) + expect(content).toContain('- Messages: 3') + }) + + /** + * `data` is untyped jsonb, so it can hold legacy or partially-written shapes. The + * filter mirrors `Memory.fetchMemory` so a transcript never contains anything the + * agent itself would not read back. + */ + it('drops entries the agent would not read back', () => { + const content = renderTranscript(META, [ + { role: 'user', content: 'kept' }, + { role: 'tool', content: 'unknown role' }, + { role: 'assistant', content: 42 }, + { role: 'assistant' }, + null, + 'not an object', + ]) + + expect(content).toContain('kept') + expect(content).not.toContain('unknown role') + expect(content).not.toContain('42') + expect(content).toContain('- Messages: 1') + }) + + it('tolerates a non-array data column', () => { + for (const value of [null, undefined, {}, 'oops', 7]) { + const content = renderTranscript(META, value) + expect(content).toContain('- Messages: 0') + } + }) + + it('normalizes CRLF and trims each message', () => { + const content = renderTranscript(META, [{ role: 'user', content: ' line1\r\nline2 ' }]) + expect(content).toContain('line1\nline2') + expect(content).not.toContain('\r') + }) + + /** Recent turns are what an agent owner is analyzing, so truncation keeps the tail. */ + it('keeps the most recent messages when truncating and says so', () => { + const messages = Array.from({ length: 5_010 }, (_, index) => ({ + role: 'user', + content: `message-${index}`, + })) + + const content = renderTranscript(META, messages) + + expect(content).toContain('- Messages: 5010') + expect(content).toContain('only the most recent 5000 messages are indexed') + expect(content).toContain('message-5009') + expect(content).not.toContain('message-0\n') + }) + + it('omits the truncation note when nothing was dropped', () => { + const content = renderTranscript(META, [{ role: 'user', content: 'short' }]) + expect(content).not.toContain('only the most recent') + }) +}) + +describe('conversationToStub', () => { + it('defers content and carries tag metadata', () => { + const stub = conversationToStub(BASE_ROW) + + expect(stub.externalId).toBe('mem-1') + expect(stub.title).toBe('Conversation: support-123') + expect(stub.contentDeferred).toBe(true) + expect(stub.sourceUrl).toBeUndefined() + expect(stub.metadata).toMatchObject({ + conversationId: 'support-123', + messageCount: 6, + fileSize: 1024, + }) + }) + + it('hashes on the update watermark, which moves whenever a message is appended', () => { + const base = conversationToStub(BASE_ROW).contentHash + expect(base).toBe('memory:mem-1:2026-01-02T00:00:00.000Z:d41d8cd98f00b204e9800998ecf8427e') + + const appended = conversationToStub({ + ...BASE_ROW, + updatedAt: new Date('2026-01-03T00:00:00.000Z'), + messageCount: 8, + }) + expect(appended.contentHash).not.toBe(base) + }) + + it('is deterministic for the same row', () => { + expect(conversationToStub(BASE_ROW).contentHash).toBe( + conversationToStub({ ...BASE_ROW }).contentHash + ) + }) + + /** + * `updatedAt` is only millisecond-resolution, so appends landing in the same + * millisecond as the indexed value would otherwise hash identically and the sync + * engine would call the transcript unchanged — leaving the new messages out of + * the knowledge base until some later write moved the clock. + */ + it('distinguishes appends that share a millisecond with the indexed value', () => { + const indexed = conversationToStub(BASE_ROW).contentHash + const appendedSameMs = conversationToStub({ + ...BASE_ROW, + messageCount: BASE_ROW.messageCount + 2, + contentDigest: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }).contentHash + + expect(appendedSameMs).not.toBe(indexed) + }) + + /** + * The case metadata proxies could not close: a same-millisecond replacement that + * preserves both message count and stored byte size. Only the content digest moves. + */ + it('distinguishes a replacement that preserves count and byte size', () => { + expect( + conversationToStub({ + ...BASE_ROW, + contentDigest: 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + }).contentHash + ).not.toBe(conversationToStub(BASE_ROW).contentHash) + }) + + /** Metadata churn without a content change must NOT force a re-index. */ + it('is unchanged when only the byte-size hint moves', () => { + expect( + conversationToStub({ ...BASE_ROW, approxBytes: BASE_ROW.approxBytes + 40 }).contentHash + ).toBe(conversationToStub(BASE_ROW).contentHash) + }) +}) + +describe('cursor', () => { + it('round-trips a keyset position', () => { + const row = { updatedAt: new Date('2026-01-02T03:04:05.678Z'), id: 'mem-9' } + const decoded = decodeCursor(encodeCursor(row)) + + expect(decoded.id).toBe('mem-9') + expect(decoded.updatedAt.toISOString()).toBe(row.updatedAt.toISOString()) + }) + + it('throws on a malformed cursor rather than restarting the listing', () => { + expect(() => decodeCursor('nonsense')).toThrow(/Malformed/) + expect(() => decodeCursor('2026-01-01T00:00:00.000Z|')).toThrow(/Malformed/) + }) +}) + +describe('parseOptionalPositiveInt', () => { + it('treats blank values as unset and rejects non-positive integers', () => { + expect(parseOptionalPositiveInt('')).toBeUndefined() + expect(parseOptionalPositiveInt('2')).toBe(2) + expect(parseOptionalPositiveInt('0')).toBeNull() + expect(parseOptionalPositiveInt('1.5')).toBeNull() + }) +}) diff --git a/apps/sim/connectors/sim-conversations/sim-conversations.ts b/apps/sim/connectors/sim-conversations/sim-conversations.ts new file mode 100644 index 00000000000..64560cc5cb8 --- /dev/null +++ b/apps/sim/connectors/sim-conversations/sim-conversations.ts @@ -0,0 +1,440 @@ +import { db } from '@sim/db' +import { memory, memorySecretProvenance } from '@sim/db/schema' +import { and, asc, desc, eq, gt, isNull, lt, or, type SQL, sql } from 'drizzle-orm' +import { readBoundMemorySecretProvenance } from '@/lib/memory/secret-provenance' +import { simConversationsConnectorMeta } from '@/connectors/sim-conversations/meta' +import type { + ConnectorConfig, + ConnectorSyncContext, + ExternalDocument, + ExternalDocumentList, +} from '@/connectors/types' +import { + CONNECTOR_MAX_FILE_BYTES, + isSkippedDocument, + markSkipped, + parseTagDate, + sizeLimitSkipReason, + takeIndexableWithinCap, +} from '@/connectors/utils' +import { MEMORY } from '@/executor/constants' + +const PAGE_SIZE = 100 + +/** + * Upper bound on messages rendered into one transcript. A conversation is appended to + * indefinitely (nothing expires an active `memory` row), so without a bound a + * long-running agent would eventually produce a document too large to embed. + */ +const MAX_TRANSCRIPT_MESSAGES = 5_000 + +/** Roles the agent itself would read back. Mirrors `Memory.fetchMemory`. */ +const INDEXABLE_ROLES = new Set(['system', 'user', 'assistant']) + +export interface ConversationRow { + id: string + key: string + createdAt: Date + updatedAt: Date + messageCount: number + approxBytes: number + contentDigest: string +} + +interface Cursor { + updatedAt: Date + id: string +} + +export function encodeCursor(row: { updatedAt: Date; id: string }): string { + return `${row.updatedAt.toISOString()}|${row.id}` +} + +/** Throws rather than rewinding — see the matching helper in the files connector. */ +export function decodeCursor(cursor: string): Cursor { + const separator = cursor.indexOf('|') + const updatedAt = separator === -1 ? Number.NaN : Date.parse(cursor.slice(0, separator)) + const id = separator === -1 ? '' : cursor.slice(separator + 1) + if (Number.isNaN(updatedAt) || !id) { + throw new Error(`Malformed conversation cursor: ${cursor}`) + } + return { updatedAt: new Date(updatedAt), id } +} + +/** + * Escapes LIKE metacharacters so a prefix is matched literally. + * + * Without this a prefix of `%` matches every conversation in the workspace and `_` + * matches any single character — turning a narrow filter into a full export. + * Must be paired with `ESCAPE '\'`. + */ +export function escapeLikePrefix(prefix: string): string { + return prefix.replace(/[\\%_]/g, (char) => `\\${char}`) +} + +/** Parses an optional positive-integer config field. `null` signals invalid input. */ +export function parseOptionalPositiveInt(value: unknown): number | null | undefined { + if (value === undefined || value === null || value === '') return undefined + const parsed = Number(value) + if (!Number.isInteger(parsed) || parsed <= 0) return null + return parsed +} + +/** + * The full `where` for a listing page. + * + * `workspaceId` is the engine-supplied one and is applied unconditionally, so a + * `sourceConfig` carrying its own cannot widen the query. `deletedAt IS NULL` is + * required and is notably *absent* from `Memory.fetchMemory`; do not copy that. + * Kept pure and exported so both invariants are directly testable. + */ +export function buildConversationListingFilters(args: { + workspaceId: string + prefix: string + cursor?: Cursor + /** Must match the query's ORDER BY, or the keyset walks the wrong way. */ + descending?: boolean +}): SQL[] { + const filters: SQL[] = [eq(memory.workspaceId, args.workspaceId), isNull(memory.deletedAt)] + + if (args.prefix) { + filters.push(sql`${memory.key} LIKE ${`${escapeLikePrefix(args.prefix)}%`} ESCAPE '\\'` as SQL) + } + + if (args.cursor) { + const beyond = args.descending ? lt : gt + filters.push( + or( + beyond(memory.updatedAt, args.cursor.updatedAt), + and(eq(memory.updatedAt, args.cursor.updatedAt), beyond(memory.id, args.cursor.id)) + ) as SQL + ) + } + + return filters +} + +/** + * Renders a stored conversation as a markdown transcript. + * + * Message filtering mirrors `Memory.fetchMemory` (`executor/handlers/agent/memory.ts`) + * so a transcript never surfaces anything the agent itself would not read back — the + * `data` column is untyped `jsonb` and can hold partially-written or legacy shapes. + * + * Keeps the most recent messages when truncating: recent turns are what an owner + * analyzing an agent actually wants, and the header records that it happened. + */ +export function renderTranscript( + meta: { conversationId: string; startedAt: string; lastActivity: string }, + rawData: unknown +): string { + const all = Array.isArray(rawData) ? rawData : [] + const messages = all.filter( + (entry): entry is { role: string; content: string } => + Boolean(entry) && + typeof entry === 'object' && + typeof (entry as { role?: unknown }).role === 'string' && + INDEXABLE_ROLES.has((entry as { role: string }).role) && + typeof (entry as { content?: unknown }).content === 'string' + ) + + const truncated = messages.length > MAX_TRANSCRIPT_MESSAGES + const kept = truncated ? messages.slice(-MAX_TRANSCRIPT_MESSAGES) : messages + + const header = [ + `# Conversation: ${meta.conversationId}`, + '', + `- Conversation ID: ${meta.conversationId}`, + `- Messages: ${messages.length}`, + `- Started: ${meta.startedAt}`, + `- Last activity: ${meta.lastActivity}`, + ...(truncated + ? [`- Note: only the most recent ${MAX_TRANSCRIPT_MESSAGES} messages are indexed.`] + : []), + '', + '---', + '', + ] + + const body = kept.map( + (message) => + `## ${message.role.charAt(0).toUpperCase()}${message.role.slice(1)}\n\n${message.content.replace(/\r\n/g, '\n').trim()}\n` + ) + + return [...header, ...body].join('\n') +} + +/** + * Builds the listing stub for one conversation. + * + * Single source of truth for `contentHash`, used by both listing and hydration. + * + * Keyed on a digest of the stored JSON rather than on metadata proxies. `updatedAt` + * is only millisecond-resolution, and message count plus byte size still collide for + * a same-millisecond replacement that happens to preserve both — each proxy narrows + * the window without closing it. A content digest closes it outright: the hash moves + * if and only if the transcript moved. + * + * Postgres computes the digest, so the listing transfers 32 characters instead of the + * payload — the reason `data` is deliberately not selected here. `updatedAt` stays in + * the hash purely so a stored value is legible when debugging. + */ +export function conversationToStub(row: ConversationRow): ExternalDocument { + return { + externalId: row.id, + title: `Conversation: ${row.key}`, + content: '', + contentDeferred: true, + mimeType: 'text/plain', + // No sourceUrl: conversations have no page of their own in the app. + contentHash: `memory:${row.id}:${row.updatedAt.toISOString()}:${row.contentDigest}`, + metadata: { + conversationId: row.key, + messageCount: row.messageCount, + lastActivity: row.updatedAt.toISOString(), + startedAt: row.createdAt.toISOString(), + /** + * Paces hydration against the engine's in-flight byte budget. `pg_column_size` + * reports the compressed, post-TOAST size, so this under-reports — it is a + * batching hint only, and `getDocument` makes the authoritative size decision + * against the rendered transcript. + */ + fileSize: row.approxBytes, + }, + } +} + +const CONVERSATION_ROW_COLUMNS = { + id: memory.id, + key: memory.key, + createdAt: memory.createdAt, + updatedAt: memory.updatedAt, + /** + * Never select `data` while listing — the transcript is the entire payload, and a + * page of them would dwarf the metadata this phase actually needs. `jsonb_typeof` + * guards the untyped column: `jsonb_array_length` errors on a non-array. + */ + messageCount: sql` + CASE WHEN jsonb_typeof(${memory.data}) = 'array' + THEN jsonb_array_length(${memory.data}) ELSE 0 END`, + approxBytes: sql`pg_column_size(${memory.data})`, + /** + * Content-addressed change detection, computed in Postgres so the transcript + * itself never crosses the wire during listing. `jsonb::text` is canonical + * (keys sorted, whitespace normalized), so the digest is stable for a value. + */ + contentDigest: sql`md5(${memory.data}::text)`, +} as const + +function readPrefix(sourceConfig: Record): string { + return typeof sourceConfig.conversationIdPrefix === 'string' + ? sourceConfig.conversationIdPrefix.trim() + : '' +} + +export const simConversationsConnector: ConnectorConfig = { + ...simConversationsConnectorMeta, + + listDocuments: async ( + _accessToken: string, + sourceConfig: Record, + cursor: string | undefined, + syncContext: ConnectorSyncContext + ): Promise => { + const workspaceId = syncContext.workspaceId + const minMessages = parseOptionalPositiveInt(sourceConfig.minMessages) ?? 1 + const maxConversations = parseOptionalPositiveInt(sourceConfig.maxConversations) ?? 0 + + /** + * See the matching note in the files connector: ascending is the safe walk for a + * complete listing, but under a cap it would mean "the oldest N" and would freeze + * an already-indexed conversation the moment it received a new message. + */ + const descending = maxConversations > 0 + + const rows = await db + .select(CONVERSATION_ROW_COLUMNS) + .from(memory) + .where( + and( + ...buildConversationListingFilters({ + workspaceId, + prefix: readPrefix(sourceConfig), + cursor: cursor ? decodeCursor(cursor) : undefined, + descending, + }) + ) + ) + .orderBy( + descending ? desc(memory.updatedAt) : asc(memory.updatedAt), + descending ? desc(memory.id) : asc(memory.id) + ) + .limit(PAGE_SIZE) + + const items: ExternalDocument[] = [] + for (const row of rows) { + // A deliberate scope filter, so it must not set `listingCapped` — a conversation + // that drops below the threshold should reconcile away like any other removal. + if (row.messageCount < minMessages) continue + items.push(conversationToStub(row)) + } + + const lastRow = rows.at(-1) + const pageFilled = rows.length === PAGE_SIZE + + const indexedSoFar = (syncContext.simConversationsIndexed as number | undefined) ?? 0 + const { documents, indexableCount, capReached } = takeIndexableWithinCap( + items, + isSkippedDocument, + maxConversations, + indexedSoFar + ) + syncContext.simConversationsIndexed = indexedSoFar + indexableCount + + if (capReached) { + /** + * Only a genuinely truncated listing blocks deletion reconciliation — see the + * matching note in the files connector. Exhausting the source at exactly + * `maxConversations` is a complete listing. + */ + /** + * A capped listing can NEVER certify completeness, so it always blocks deletion + * reconciliation — even when the source happened to run out at exactly + * `maxConversations` with nothing dropped. + * + * The reason is the descending order a cap implies: a row updated between pages + * moves ABOVE the keyset cursor and is skipped for this run. (Ascending has the + * opposite skew — the row moves below the cursor and is re-seen, which the + * engine dedupes.) So "we consumed exactly the budget" does not prove "we saw + * everything", and treating it as proof lets reconciliation hard-delete a source + * item that still exists. + * + * Do not relax this into a `droppedFromPage || morePagesAvailable` check. That + * reads correct in isolation and is how the Asana connector decides truncation, + * but Asana lists ascending — the inference does not transfer. + */ + syncContext.listingCapped = true + return { documents, hasMore: false } + } + + return { + documents, + nextCursor: pageFilled && lastRow ? encodeCursor(lastRow) : undefined, + hasMore: pageFilled, + } + }, + + getDocument: async ( + _accessToken: string, + _sourceConfig: Record, + externalId: string, + syncContext: ConnectorSyncContext + ): Promise => { + const workspaceId = syncContext.workspaceId + + // Re-read through the same predicates: never fetch by external id alone. + // Left-joins the provenance sidecar so the secret check below has its inputs. + const rows = await db + .select({ + ...CONVERSATION_ROW_COLUMNS, + data: memory.data, + secretProvenanceVersion: memory.secretProvenanceVersion, + provenanceContentHash: memorySecretProvenance.contentHash, + provenanceStatus: memorySecretProvenance.status, + provenanceEntries: memorySecretProvenance.entries, + }) + .from(memory) + .leftJoin(memorySecretProvenance, eq(memorySecretProvenance.memoryId, memory.id)) + .where( + and( + eq(memory.id, externalId), + eq(memory.workspaceId, workspaceId), + isNull(memory.deletedAt) + ) + ) + .limit(1) + + const row = rows[0] + if (!row) return null + + const stub = conversationToStub(row) + + /** + * Agent memory is where resolved credentials and env values land in message + * text, so every other reader of `memory.data` pairs it with this sidecar + * (see `app/api/memory/route.ts`). Indexing a transcript copies it into KB + * chunks and embeddings, which are readable by anyone with *any* permission on + * the workspace — a wider audience than the write/admin needed to create the + * connector — so only a provably secret-free conversation is indexed. + * + * `readBoundMemorySecretProvenance` returns exact-empty for untracked legacy + * rows and `unknown` for malformed ones, so this fails closed. + */ + const provenance = readBoundMemorySecretProvenance({ + secretProvenanceVersion: row.secretProvenanceVersion, + data: row.data, + provenanceContentHash: row.provenanceContentHash, + status: row.provenanceStatus, + entries: row.provenanceEntries, + }) + if (provenance.status !== 'exact' || provenance.entries.length > 0) { + return markSkipped( + stub, + 'Conversation contains secret-derived values or its provenance is unavailable, so it was not indexed' + ) + } + + const content = renderTranscript( + { + conversationId: row.key, + startedAt: row.createdAt.toISOString(), + lastActivity: row.updatedAt.toISOString(), + }, + row.data + ) + + if (Buffer.byteLength(content, 'utf8') > CONNECTOR_MAX_FILE_BYTES) { + return markSkipped(stub, sizeLimitSkipReason(CONNECTOR_MAX_FILE_BYTES)) + } + if (!content.trim()) return null + + return { ...stub, content, contentDeferred: false } + }, + + validateConfig: async ( + _accessToken: string, + sourceConfig: Record + ): Promise<{ valid: boolean; error?: string }> => { + const prefix = readPrefix(sourceConfig) + if (prefix.length > MEMORY.MAX_CONVERSATION_ID_LENGTH) { + return { + valid: false, + error: `Conversation ID Prefix cannot exceed ${MEMORY.MAX_CONVERSATION_ID_LENGTH} characters`, + } + } + + if (parseOptionalPositiveInt(sourceConfig.minMessages) === null) { + return { valid: false, error: 'Minimum Messages must be a positive whole number' } + } + if (parseOptionalPositiveInt(sourceConfig.maxConversations) === null) { + return { valid: false, error: 'Max Conversations must be a positive whole number' } + } + + return { valid: true } + }, + + mapTags: (metadata: Record): Record => { + const tags: Record = {} + + if (typeof metadata.conversationId === 'string' && metadata.conversationId) { + tags.conversationId = metadata.conversationId + } + if (typeof metadata.messageCount === 'number' && Number.isFinite(metadata.messageCount)) { + tags.messageCount = metadata.messageCount + } + const lastActivity = parseTagDate(metadata.lastActivity) + if (lastActivity) tags.lastActivity = lastActivity + + return tags + }, +} diff --git a/apps/sim/connectors/sim-files/index.ts b/apps/sim/connectors/sim-files/index.ts new file mode 100644 index 00000000000..122c2d8ae0b --- /dev/null +++ b/apps/sim/connectors/sim-files/index.ts @@ -0,0 +1 @@ +export { simFilesConnector } from '@/connectors/sim-files/sim-files' diff --git a/apps/sim/connectors/sim-files/meta.ts b/apps/sim/connectors/sim-files/meta.ts new file mode 100644 index 00000000000..854959447a3 --- /dev/null +++ b/apps/sim/connectors/sim-files/meta.ts @@ -0,0 +1,82 @@ +import { SimLogoIcon } from '@/components/icons' +import type { ConnectorMeta } from '@/connectors/types' + +export const simFilesConnectorMeta: ConnectorMeta = { + id: 'sim_files', + name: 'Workspace Files', + description: 'Sync files from this workspace so agents can search their contents', + version: '1.0.0', + icon: SimLogoIcon, + + auth: { mode: 'sim' }, + + configFields: [ + { + id: 'folderSelector', + title: 'Folder', + type: 'selector', + selectorKey: 'sim.fileFolders', + canonicalParamId: 'folderId', + mode: 'basic', + required: false, + placeholder: 'All files', + description: 'Limit the sync to one folder. Leave empty to sync every file in the workspace.', + }, + { + id: 'folderId', + title: 'Folder ID', + type: 'short-input', + canonicalParamId: 'folderId', + mode: 'advanced', + required: false, + placeholder: 'e.g. 8f2c4d1e-…', + }, + { + id: 'recursive', + title: 'Include Subfolders', + type: 'dropdown', + required: false, + options: [ + { label: 'Yes', id: 'true' }, + { label: 'No', id: 'false' }, + ], + description: 'Defaults to Yes.', + }, + { + id: 'extensions', + title: 'File Types', + type: 'short-input', + multi: true, + required: false, + placeholder: 'e.g. pdf, docx, md', + description: + 'Comma-separated extensions. Leave empty to sync every readable type. Files Sim cannot extract text from (images, archives, audio) are never synced.', + }, + { + id: 'maxFiles', + title: 'Max Files', + type: 'short-input', + required: false, + placeholder: 'e.g. 1000', + description: 'Caps how many files are indexed. Leave empty for no limit.', + }, + ], + + /** + * Deliberately absent. Listing here is one indexed query against a local table, so + * the usual reason to sync incrementally (an external API's rate limit) does not + * apply — while `shouldReconcileDeletions` disables deletion reconciliation for any + * incremental run, and `contentUpdatedAt` advances only on content writes, so + * renames, moves and deletes would never reach the knowledge base. Content is still + * only re-fetched when a document's `contentHash` actually changes. + */ + supportsIncrementalSync: false, + + tagDefinitions: [ + { id: 'folderPath', displayName: 'Folder Path', fieldType: 'text' }, + { id: 'contentType', displayName: 'Content Type', fieldType: 'text' }, + { id: 'uploadedBy', displayName: 'Uploaded By', fieldType: 'text' }, + { id: 'fileSize', displayName: 'File Size', fieldType: 'number' }, + { id: 'lastModified', displayName: 'Last Modified', fieldType: 'date' }, + ], +} diff --git a/apps/sim/connectors/sim-files/sim-files.test.ts b/apps/sim/connectors/sim-files/sim-files.test.ts new file mode 100644 index 00000000000..d33ddf743d8 --- /dev/null +++ b/apps/sim/connectors/sim-files/sim-files.test.ts @@ -0,0 +1,293 @@ +/** + * @vitest-environment node + */ +import { flattenMockConditions, type MockCondition } from '@sim/testing' +import { describe, expect, it } from 'vitest' +import { + buildFileListingFilters, + collectDescendantFolderIds, + decodeCursor, + encodeCursor, + type FileRow, + fileRowToStub, + normalizeExt, + parseOptionalPositiveInt, +} from '@/connectors/sim-files/sim-files' + +const BASE_ROW: FileRow = { + id: 'file-1', + originalName: 'spec.md', + contentType: 'text/markdown', + size: 2048, + folderId: 'folder-specs', + userId: 'user-1', + contentUpdatedAt: new Date('2026-01-01T00:00:00.000Z'), + updatedAt: new Date('2026-01-01T00:00:00.000Z'), +} + +function conditionsOf(args: Parameters[0]): MockCondition[] { + return buildFileListingFilters(args).flatMap(flattenMockConditions) +} + +/** The schema mock represents each column as its own camelCase name. */ +function columnNames(nodes: MockCondition[]): string[] { + return nodes.map((node) => String(node.left ?? node.column ?? '')) +} + +/** `flattenMockConditions` unwraps `and` but not `or`, which the keyset clause uses. */ +function orBranches(nodes: MockCondition[]): MockCondition[] { + return nodes + .filter((node) => node.type === 'or') + .flatMap((node) => (node.conditions as MockCondition[]) ?? []) + .flatMap(flattenMockConditions) +} + +describe('collectDescendantFolderIds', () => { + const FOLDERS = [ + { id: 'root', parentId: null }, + { id: 'docs', parentId: 'root' }, + { id: 'specs', parentId: 'docs' }, + { id: 'deep', parentId: 'specs' }, + { id: 'other', parentId: 'root' }, + ] + + it('collects the whole subtree including the root itself', () => { + expect(collectDescendantFolderIds(FOLDERS, 'docs').sort()).toEqual(['deep', 'docs', 'specs']) + }) + + it('excludes siblings and ancestors', () => { + const result = collectDescendantFolderIds(FOLDERS, 'specs') + expect(result.sort()).toEqual(['deep', 'specs']) + expect(result).not.toContain('other') + expect(result).not.toContain('docs') + }) + + it('returns just the root when it has no children', () => { + expect(collectDescendantFolderIds(FOLDERS, 'deep')).toEqual(['deep']) + }) + + it('returns the id itself when the folder is unknown', () => { + expect(collectDescendantFolderIds(FOLDERS, 'missing')).toEqual(['missing']) + }) + + /** An unguarded breadth-first walk would spin forever on this input. */ + it('terminates on a parentId cycle', () => { + const cyclic = [ + { id: 'a', parentId: 'b' }, + { id: 'b', parentId: 'a' }, + ] + expect(collectDescendantFolderIds(cyclic, 'a').sort()).toEqual(['a', 'b']) + }) +}) + +describe('fileRowToStub', () => { + it('defers content and reports size for the engine byte budget', () => { + const stub = fileRowToStub(BASE_ROW, 'ws-1', 'Docs/Specs') + + expect(stub.externalId).toBe('file-1') + expect(stub.title).toBe('spec.md') + expect(stub.contentDeferred).toBe(true) + expect(stub.content).toBe('') + expect(stub.metadata?.fileSize).toBe(2048) + expect(stub.metadata?.folderPath).toBe('Docs/Specs') + }) + + /** + * A rename or a move changes what gets indexed (title, folder tag) without touching + * `contentUpdatedAt`, so the hash must move or the knowledge base keeps stale values. + */ + it('changes the hash on rename, move, and content write', () => { + const base = fileRowToStub(BASE_ROW, 'ws-1', 'Docs/Specs').contentHash + + const renamed = fileRowToStub({ ...BASE_ROW, originalName: 'spec-v2.md' }, 'ws-1', 'Docs/Specs') + const moved = fileRowToStub({ ...BASE_ROW, folderId: 'folder-other' }, 'ws-1', 'Docs/Other') + const rewritten = fileRowToStub( + { ...BASE_ROW, contentUpdatedAt: new Date('2026-02-02T00:00:00.000Z') }, + 'ws-1', + 'Docs/Specs' + ) + + expect(renamed.contentHash).not.toBe(base) + expect(moved.contentHash).not.toBe(base) + expect(rewritten.contentHash).not.toBe(base) + }) + + /** + * `updatedAt` moves on metadata-only writes. Re-hashing on those would re-download + * and re-embed every file for no reason. + */ + it('keeps the hash stable when only updatedAt moves', () => { + const before = fileRowToStub(BASE_ROW, 'ws-1', 'Docs/Specs').contentHash + const after = fileRowToStub( + { ...BASE_ROW, updatedAt: new Date('2026-03-03T00:00:00.000Z') }, + 'ws-1', + 'Docs/Specs' + ).contentHash + + expect(after).toBe(before) + }) + + /** + * Renaming an ancestor folder rewrites the stored path tag but writes only the + * `folder` table — `contentUpdatedAt`, `originalName` and `folderId` all stay put, + * so without the path in the hash the document keeps the old folder tag forever. + */ + it('changes the hash when an ancestor folder rename moves the path', () => { + expect(fileRowToStub(BASE_ROW, 'ws-1', 'Documentation/Specs').contentHash).not.toBe( + fileRowToStub(BASE_ROW, 'ws-1', 'Docs/Specs').contentHash + ) + }) + + /** listDocuments and getDocument must produce byte-identical hashes for one row. */ + it('is deterministic for the same row', () => { + expect(fileRowToStub(BASE_ROW, 'ws-1', 'Docs/Specs').contentHash).toBe( + fileRowToStub({ ...BASE_ROW }, 'ws-1', 'Docs/Specs').contentHash + ) + }) +}) + +describe('buildFileListingFilters', () => { + /** + * The builder takes no `sourceConfig`, so this proves only that the supplied + * workspace is bound. That the connector never READS `sourceConfig.workspaceId` + * is proven end to end by the sync harness, not here. + */ + it('binds the supplied workspace', () => { + const nodes = conditionsOf({ workspaceId: 'ws-real', folderIds: null, rootOnly: false }) + const workspaceClause = nodes.find((node) => node.left === 'workspaceId') + + expect(workspaceClause).toMatchObject({ type: 'eq', right: 'ws-real' }) + }) + + /** + * Without the `context` clause a knowledge base would ingest every member's private + * copilot and chat attachments, which live in the same table under the same workspace. + */ + it('always restricts to workspace-context files and excludes deleted rows', () => { + const nodes = conditionsOf({ workspaceId: 'ws-1', folderIds: null, rootOnly: false }) + + expect(nodes).toContainEqual(expect.objectContaining({ type: 'eq', right: 'workspace' })) + expect(columnNames(nodes)).toContain('deletedAt') + }) + + it('scopes to a folder subtree when one is given', () => { + const nodes = conditionsOf({ + workspaceId: 'ws-1', + folderIds: ['folder-a', 'folder-b'], + rootOnly: false, + }) + + expect(nodes).toContainEqual( + expect.objectContaining({ type: 'inArray', values: ['folder-a', 'folder-b'] }) + ) + }) + + /** `inArray(col, [])` is invalid SQL, so an empty scope needs its own match-nothing clause. */ + it('matches nothing rather than everything for an empty folder scope', () => { + const nodes = conditionsOf({ workspaceId: 'ws-1', folderIds: [], rootOnly: false }) + + expect(nodes).toContainEqual(expect.objectContaining({ type: 'eq', right: '' })) + expect(nodes.some((node) => node.type === 'inArray')).toBe(false) + }) + + it('restricts to unfoldered files when subfolders are excluded at the root', () => { + const nodes = conditionsOf({ workspaceId: 'ws-1', folderIds: null, rootOnly: true }) + expect(columnNames(nodes.filter((n) => n.type === 'isNull'))).toContain('folderId') + }) + + /** + * The keyset direction must match ORDER BY. Ascending walks forward with `gt`; + * descending (used when a cap is set, so the cap means "most recently active N" + * rather than "oldest N") must walk with `lt` or the second page repeats page one. + */ + it('flips the keyset comparison when the listing is descending', () => { + const cursor = { updatedAt: new Date('2026-01-01T00:00:00.000Z'), id: 'file-1' } + const ascending = orBranches( + conditionsOf({ workspaceId: 'ws-1', folderIds: null, rootOnly: false, cursor }) + ) + const descending = orBranches( + conditionsOf({ + workspaceId: 'ws-1', + folderIds: null, + rootOnly: false, + cursor, + descending: true, + }) + ) + + expect(ascending.some((node) => node.type === 'gt')).toBe(true) + expect(ascending.some((node) => node.type === 'lt')).toBe(false) + expect(descending.some((node) => node.type === 'lt')).toBe(true) + expect(descending.some((node) => node.type === 'gt')).toBe(false) + }) + + it('adds a keyset clause only when paginating', () => { + const first = conditionsOf({ workspaceId: 'ws-1', folderIds: null, rootOnly: false }) + const next = conditionsOf({ + workspaceId: 'ws-1', + folderIds: null, + rootOnly: false, + cursor: { updatedAt: new Date('2026-01-01T00:00:00.000Z'), id: 'file-1' }, + }) + + expect(first.some((node) => node.type === 'or')).toBe(false) + expect(orBranches(next).some((node) => node.type === 'gt')).toBe(true) + expect(columnNames(orBranches(next))).toContain('updatedAt') + }) +}) + +describe('cursor', () => { + it('round-trips a keyset position', () => { + const row = { updatedAt: new Date('2026-01-02T03:04:05.678Z'), id: 'file-9' } + const decoded = decodeCursor(encodeCursor(row)) + + expect(decoded.id).toBe('file-9') + expect(decoded.updatedAt.toISOString()).toBe(row.updatedAt.toISOString()) + }) + + it('preserves ids containing the separator', () => { + const row = { updatedAt: new Date('2026-01-02T03:04:05.678Z'), id: 'weird|id' } + expect(decodeCursor(encodeCursor(row)).id).toBe('weird|id') + }) + + /** Rewinding instead would silently re-emit page one until MAX_PAGES. */ + it('throws on a malformed cursor rather than restarting the listing', () => { + expect(() => decodeCursor('nonsense')).toThrow(/Malformed/) + expect(() => decodeCursor('not-a-date|file-1')).toThrow(/Malformed/) + expect(() => decodeCursor('2026-01-01T00:00:00.000Z|')).toThrow(/Malformed/) + }) +}) + +describe('normalizeExt', () => { + it.each([ + ['spec.md', 'md'], + ['REPORT.PDF', 'pdf'], + ['archive.tar.gz', 'gz'], + [' notes.TXT ', 'txt'], + ['pdf', 'pdf'], + ['.pdf', 'pdf'], + ['Makefile', 'makefile'], + ])('normalizes %s to %s', (input, expected) => { + expect(normalizeExt(input)).toBe(expected) + }) +}) + +describe('parseOptionalPositiveInt', () => { + it('treats blank values as unset', () => { + expect(parseOptionalPositiveInt(undefined)).toBeUndefined() + expect(parseOptionalPositiveInt('')).toBeUndefined() + expect(parseOptionalPositiveInt(null)).toBeUndefined() + }) + + it('accepts positive integers', () => { + expect(parseOptionalPositiveInt('1000')).toBe(1000) + expect(parseOptionalPositiveInt(5)).toBe(5) + }) + + it('rejects zero, negatives, fractions, and non-numbers', () => { + expect(parseOptionalPositiveInt('0')).toBeNull() + expect(parseOptionalPositiveInt('-3')).toBeNull() + expect(parseOptionalPositiveInt('1.5')).toBeNull() + expect(parseOptionalPositiveInt('lots')).toBeNull() + }) +}) diff --git a/apps/sim/connectors/sim-files/sim-files.ts b/apps/sim/connectors/sim-files/sim-files.ts new file mode 100644 index 00000000000..5a7c5b61bb0 --- /dev/null +++ b/apps/sim/connectors/sim-files/sim-files.ts @@ -0,0 +1,579 @@ +import { db } from '@sim/db' +import { workspaceFiles } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import { and, asc, desc, eq, gt, inArray, isNull, lt, or, type SQL } from 'drizzle-orm' +import { getBaseUrl } from '@/lib/core/utils/urls' +import { isSupportedFileType, parseBuffer } from '@/lib/file-parsers' +import { listWorkspaceFileFolders } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager' +import { + fetchServableWorkspaceFileBuffer, + getWorkspaceFile, +} from '@/lib/uploads/contexts/workspace/workspace-file-manager' +import { + isModelSafeWorkspaceFileKey, + MODEL_UNSAFE_WORKSPACE_FILE_ERROR_MESSAGE, +} from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance' +import { simFilesConnectorMeta } from '@/connectors/sim-files/meta' +import type { + ConnectorConfig, + ConnectorSyncContext, + ExternalDocument, + ExternalDocumentList, +} from '@/connectors/types' +import { + CONNECTOR_MAX_FILE_BYTES, + isSkippedDocument, + markSkipped, + parseMultiValue, + parseTagDate, + sizeLimitSkipReason, + stubOrSkipBySize, + takeIndexableWithinCap, +} from '@/connectors/utils' + +const logger = createLogger('SimFilesConnector') + +const PAGE_SIZE = 200 + +/** + * Only files the workspace's own Files module owns. The `workspace_files` table is + * multi-tenant by `context` as well as by workspace: `mothership`/`copilot`/`chat`/ + * `execution` rows are per-user attachments that still carry a `workspaceId`, so + * dropping this predicate would sync every member's private uploads into a shared + * knowledge base. + */ +const WORKSPACE_FILE_CONTEXT = 'workspace' + +/** Columns the stub is built from. Selected identically by listing and hydration. */ +const FILE_ROW_COLUMNS = { + id: workspaceFiles.id, + originalName: workspaceFiles.originalName, + contentType: workspaceFiles.contentType, + size: workspaceFiles.size, + folderId: workspaceFiles.folderId, + userId: workspaceFiles.userId, + contentUpdatedAt: workspaceFiles.contentUpdatedAt, + updatedAt: workspaceFiles.updatedAt, +} as const + +export interface FileRow { + id: string + originalName: string + contentType: string + size: number + folderId: string | null + userId: string + contentUpdatedAt: Date + updatedAt: Date +} + +/** Keyset position: the last row emitted, ordered by `(updatedAt, id)`. */ +interface Cursor { + updatedAt: Date + id: string +} + +export function encodeCursor(row: { updatedAt: Date; id: string }): string { + return `${row.updatedAt.toISOString()}|${row.id}` +} + +/** + * Throws rather than restarting on a malformed cursor: the engine only ever hands + * back a cursor this connector produced, so a bad one means a real bug, and silently + * rewinding would re-emit page one forever. + */ +export function decodeCursor(cursor: string): Cursor { + const separator = cursor.indexOf('|') + const updatedAt = separator === -1 ? Number.NaN : Date.parse(cursor.slice(0, separator)) + const id = separator === -1 ? '' : cursor.slice(separator + 1) + if (Number.isNaN(updatedAt) || !id) { + throw new Error(`Malformed workspace files cursor: ${cursor}`) + } + return { updatedAt: new Date(updatedAt), id } +} + +/** Lowercased, dot-stripped extension of a filename, or '' when it has none. */ +export function normalizeExt(value: string): string { + const trimmed = value.trim().toLowerCase() + const dot = trimmed.lastIndexOf('.') + return dot === -1 ? trimmed : trimmed.slice(dot + 1) +} + +/** + * Every folder at or beneath `rootId`, breadth-first. + * + * Guards against a cycle in `parentId`: the schema permits one (the DB trigger only + * enforces matching `resourceType`), and an unguarded walk would hang the sync. + */ +export function collectDescendantFolderIds( + folders: Array<{ id: string; parentId: string | null }>, + rootId: string +): string[] { + const childrenByParent = new Map() + for (const folder of folders) { + const siblings = childrenByParent.get(folder.parentId) + if (siblings) siblings.push(folder.id) + else childrenByParent.set(folder.parentId, [folder.id]) + } + + const collected: string[] = [] + const seen = new Set() + const queue: string[] = [rootId] + while (queue.length > 0) { + const id = queue.shift() as string + if (seen.has(id)) continue + seen.add(id) + collected.push(id) + const children = childrenByParent.get(id) + if (children) queue.push(...children) + } + return collected +} + +/** + * The full `where` for a listing page. + * + * `workspaceId` is the engine-supplied one and is applied unconditionally, so a + * `sourceConfig` carrying its own `workspaceId` cannot widen the query. Kept pure and + * exported so that invariant is directly testable. + */ +export function buildFileListingFilters(args: { + workspaceId: string + folderIds: string[] | null + rootOnly: boolean + cursor?: Cursor + /** Must match the query's ORDER BY, or the keyset walks the wrong way. */ + descending?: boolean +}): SQL[] { + const filters: SQL[] = [ + eq(workspaceFiles.workspaceId, args.workspaceId), + eq(workspaceFiles.context, WORKSPACE_FILE_CONTEXT), + isNull(workspaceFiles.deletedAt), + ] + + if (args.rootOnly) { + filters.push(isNull(workspaceFiles.folderId)) + } else if (args.folderIds) { + // An empty scope must match nothing, but `inArray(col, [])` is invalid SQL. + if (args.folderIds.length === 0) return [...filters, eq(workspaceFiles.id, '')] + filters.push(inArray(workspaceFiles.folderId, args.folderIds)) + } + + if (args.cursor) { + const beyond = args.descending ? lt : gt + filters.push( + or( + beyond(workspaceFiles.updatedAt, args.cursor.updatedAt), + and( + eq(workspaceFiles.updatedAt, args.cursor.updatedAt), + beyond(workspaceFiles.id, args.cursor.id) + ) + ) as SQL + ) + } + + return filters +} + +/** + * Builds the listing stub for one file. + * + * The single source of truth for `contentHash`, called by both `listDocuments` and + * `getDocument` so the two can never disagree — the engine compares the hash from + * hydration against the one from listing to decide whether a document changed. + * + * `originalName` and `folderId` participate because a rename or a move alters what + * we index (title, folder tag) without touching `contentUpdatedAt`, which advances + * only on content writes. + * + * `folderPath` participates for the same reason one level up: renaming an ANCESTOR + * folder rewrites the path we store as a tag but writes only the `folder` table, so + * every other hashed field stays put and the document would keep the old path + * forever. Both `listDocuments` and `getDocument` resolve it from the same + * `pathById` map, so the two phases still agree. + */ +export function fileRowToStub( + row: FileRow, + workspaceId: string, + folderPath: string | null +): ExternalDocument { + return { + externalId: row.id, + title: row.originalName, + content: '', + contentDeferred: true, + mimeType: 'text/plain', + sourceUrl: `${getBaseUrl()}/workspace/${workspaceId}/files${ + row.folderId ? `?folderId=${encodeURIComponent(row.folderId)}` : '' + }`, + contentHash: `simfile:${row.id}:${row.contentUpdatedAt.toISOString()}:${row.originalName}:${row.folderId ?? ''}:${folderPath ?? ''}`, + metadata: { + folderPath: folderPath ?? '', + contentType: row.contentType, + uploadedBy: row.userId, + /** + * Load-bearing: `estimateOpSizeBytes` reads `fileSize` to pace hydration + * against the engine's in-flight byte budget, and assumes 4MB without it — + * which would let several near-cap files materialize at once. + */ + fileSize: row.size, + lastModified: row.contentUpdatedAt.toISOString(), + }, + } +} + +/** Folder scope for this sync run, resolved once and cached on the sync context. */ +interface FolderScope { + folderIds: string[] | null + rootOnly: boolean + pathById: Map +} + +async function resolveFolderScope( + syncContext: ConnectorSyncContext, + workspaceId: string, + folderId: string, + recursive: boolean +): Promise { + const cached = syncContext.simFilesFolderScope as FolderScope | undefined + if (cached) return cached + + /** + * `scope: 'all'` so a file whose folder was trashed still resolves a path for its + * tag; whether the file itself is listed is governed by its own `deletedAt`. + */ + const folders = await listWorkspaceFileFolders(workspaceId, { scope: 'all' }) + const pathById = new Map(folders.map((folder) => [folder.id, folder.path])) + + let scope: FolderScope + if (!folderId) { + scope = { folderIds: null, rootOnly: !recursive, pathById } + } else if (!recursive) { + scope = { folderIds: [folderId], rootOnly: false, pathById } + } else { + scope = { folderIds: collectDescendantFolderIds(folders, folderId), rootOnly: false, pathById } + } + + syncContext.simFilesFolderScope = scope + return scope +} + +function readWorkspaceId(syncContext: ConnectorSyncContext): string { + return syncContext.workspaceId +} + +function readRecursive(sourceConfig: Record): boolean { + return sourceConfig.recursive !== 'false' +} + +function readAllowedExtensions(sourceConfig: Record): Set { + return new Set(parseMultiValue(sourceConfig.extensions).map(normalizeExt).filter(Boolean)) +} + +/** Parses an optional positive-integer config field. `null` signals invalid input. */ +export function parseOptionalPositiveInt(value: unknown): number | null | undefined { + if (value === undefined || value === null || value === '') return undefined + const parsed = Number(value) + if (!Number.isInteger(parsed) || parsed <= 0) return null + return parsed +} + +export const simFilesConnector: ConnectorConfig = { + ...simFilesConnectorMeta, + + listDocuments: async ( + _accessToken: string, + sourceConfig: Record, + cursor: string | undefined, + syncContext: ConnectorSyncContext + ): Promise => { + const workspaceId = readWorkspaceId(syncContext) + const folderId = typeof sourceConfig.folderId === 'string' ? sourceConfig.folderId.trim() : '' + const allowedExtensions = readAllowedExtensions(sourceConfig) + const maxFiles = parseOptionalPositiveInt(sourceConfig.maxFiles) ?? 0 + + let scope: FolderScope + try { + scope = await resolveFolderScope( + syncContext, + workspaceId, + folderId, + readRecursive(sourceConfig) + ) + } catch (error) { + /** + * Without a folder scope this page would silently widen to the whole workspace + * or narrow to nothing; either way the listing is not authoritative, so block + * deletion reconciliation before rethrowing. + */ + syncContext.listingCapped = true + throw error + } + + /** + * Ordering follows from whether this listing is complete or bounded. + * + * Uncapped, ascending is the safe walk: a row updated mid-sync moves toward the + * end and may be emitted twice, which the engine dedupes by `externalId`. + * + * Capped, ascending would mean "the oldest N" — and worse, an already-indexed + * file that is edited moves past the cap window, stops being listed, and (because + * `listingCapped` suppresses deletion) leaves a permanently stale document behind. + * Descending makes the cap mean "the N most recently active", which is what a + * `maxFiles` limit is for. Its own risk — a row updated mid-sync slipping behind + * the cursor — is already covered, since a capped listing is declared incomplete + * and the next sync picks the row up. + */ + const descending = maxFiles > 0 + + const rows = await db + .select(FILE_ROW_COLUMNS) + .from(workspaceFiles) + .where( + and( + ...buildFileListingFilters({ + workspaceId, + folderIds: scope.folderIds, + rootOnly: scope.rootOnly, + cursor: cursor ? decodeCursor(cursor) : undefined, + descending, + }) + ) + ) + .orderBy( + descending ? desc(workspaceFiles.updatedAt) : asc(workspaceFiles.updatedAt), + descending ? desc(workspaceFiles.id) : asc(workspaceFiles.id) + ) + .limit(PAGE_SIZE) + + const items: ExternalDocument[] = [] + for (const row of rows) { + const ext = normalizeExt(row.originalName) + /** + * Deliberate scope filters, not truncation — an unreadable type (image, + * archive, audio) or a type the user excluded must not set `listingCapped`, + * or a knowledge base could never reconcile away a file that stopped matching. + */ + if (!ext || !isSupportedFileType(ext)) continue + if (allowedExtensions.size > 0 && !allowedExtensions.has(ext)) continue + + const folderPath = row.folderId ? (scope.pathById.get(row.folderId) ?? null) : null + items.push( + stubOrSkipBySize( + fileRowToStub(row, workspaceId, folderPath), + row.size, + CONNECTOR_MAX_FILE_BYTES + ) + ) + } + + const lastRow = rows.at(-1) + const pageFilled = rows.length === PAGE_SIZE + + const indexedSoFar = (syncContext.simFilesIndexed as number | undefined) ?? 0 + const { documents, indexableCount, capReached } = takeIndexableWithinCap( + items, + isSkippedDocument, + maxFiles, + indexedSoFar + ) + syncContext.simFilesIndexed = indexedSoFar + indexableCount + + if (capReached) { + /** + * Hitting the cap is not the same as truncating. If the source ran out at + * exactly `maxFiles` — nothing dropped from this page and no further page — + * the listing is complete and safe to reconcile deletions against. Marking it + * capped anyway would suppress reconciliation forever, so a file deleted at the + * source would never leave the knowledge base. Mirrors `decideTaskCap`'s + * `droppedFromPage || (hitLimit && morePagesAvailable)` in the Asana connector. + */ + /** + * A capped listing can NEVER certify completeness, so it always blocks deletion + * reconciliation — even when the source happened to run out at exactly + * `maxFiles` with nothing dropped. + * + * The reason is the descending order a cap implies: a row updated between pages + * moves ABOVE the keyset cursor and is skipped for this run. (Ascending has the + * opposite skew — the row moves below the cursor and is re-seen, which the + * engine dedupes.) So "we consumed exactly the budget" does not prove "we saw + * everything", and treating it as proof lets reconciliation hard-delete a source + * item that still exists. + * + * Do not relax this into a `droppedFromPage || morePagesAvailable` check. That + * reads correct in isolation and is how the Asana connector decides truncation, + * but Asana lists ascending — the inference does not transfer. + */ + syncContext.listingCapped = true + return { documents, hasMore: false } + } + + return { + documents, + nextCursor: pageFilled && lastRow ? encodeCursor(lastRow) : undefined, + hasMore: pageFilled, + } + }, + + getDocument: async ( + _accessToken: string, + sourceConfig: Record, + externalId: string, + syncContext: ConnectorSyncContext + ): Promise => { + const workspaceId = readWorkspaceId(syncContext) + + // Re-read through the same predicates: never trust an external id on its own. + const rows = await db + .select(FILE_ROW_COLUMNS) + .from(workspaceFiles) + .where( + and( + eq(workspaceFiles.id, externalId), + eq(workspaceFiles.workspaceId, workspaceId), + eq(workspaceFiles.context, WORKSPACE_FILE_CONTEXT), + isNull(workspaceFiles.deletedAt) + ) + ) + .limit(1) + + const row = rows[0] + if (!row) return null + + const scope = await resolveFolderScope( + syncContext, + workspaceId, + typeof sourceConfig.folderId === 'string' ? sourceConfig.folderId.trim() : '', + readRecursive(sourceConfig) + ) + const folderPath = row.folderId ? (scope.pathById.get(row.folderId) ?? null) : null + const stub = fileRowToStub(row, workspaceId, folderPath) + + if (row.size > CONNECTOR_MAX_FILE_BYTES) { + return markSkipped(stub, sizeLimitSkipReason(CONNECTOR_MAX_FILE_BYTES)) + } + + const ext = normalizeExt(row.originalName) + if (!ext || !isSupportedFileType(ext)) return null + + /** + * `throwOnError` so a database fault surfaces as a failed sync. The default + * swallows it to `null`, which the engine reads as an empty re-fetch and records + * as a no-op — masking an outage as success. + */ + const fileRecord = await getWorkspaceFile(workspaceId, externalId, { throwOnError: true }) + if (!fileRecord) return null + + /** + * The same gate the manual knowledge-base upload path enforces + * (`assertDocumentFileModelSafe` in `documents/document-processor.ts`). + * + * It has to run HERE rather than being inherited: the sync engine re-uploads the + * extracted text under a fresh `kb/...` key, and that key has no `workspace_files` + * row — so the processor's own check resolves zero rows and passes vacuously. A + * file whose provenance is unknown would otherwise be laundered into embeddings. + * + * Skipped rather than dropped so it surfaces as a visible failed document. + */ + const provenanceSafe = await isModelSafeWorkspaceFileKey(fileRecord.key, { workspaceId }) + if (!provenanceSafe) { + return markSkipped(stub, MODEL_UNSAFE_WORKSPACE_FILE_ERROR_MESSAGE) + } + + let buffer: Buffer + try { + /** + * Not `fetchWorkspaceFileBuffer`: for generated docx/pptx/pdf/xlsx that returns + * the generation *source* (JavaScript/Python text) under a document filename, + * which the parser would happily mis-read. This resolves the rendered artifact, + * and throws `DocCompileUserError` while one is still compiling — a transient + * state that should simply be retried on the next sync. + */ + ;({ buffer } = await fetchServableWorkspaceFileBuffer(fileRecord, { + maxBytes: CONNECTOR_MAX_FILE_BYTES, + })) + } catch (error) { + logger.warn('Failed to download workspace file for indexing', { + externalId, + error: getErrorMessage(error), + }) + return null + } + + if (buffer.byteLength > CONNECTOR_MAX_FILE_BYTES) { + return markSkipped(stub, sizeLimitSkipReason(CONNECTOR_MAX_FILE_BYTES)) + } + + const { content } = await parseBuffer(buffer, ext) + if (!content.trim()) return null + + return { ...stub, content, contentDeferred: false } + }, + + validateConfig: async ( + _accessToken: string, + sourceConfig: Record, + context?: { workspaceId?: string; knowledgeBaseId?: string } + ): Promise<{ valid: boolean; error?: string }> => { + const recursive = sourceConfig.recursive + if ( + recursive !== undefined && + recursive !== '' && + recursive !== 'true' && + recursive !== 'false' + ) { + return { valid: false, error: 'Include Subfolders must be Yes or No' } + } + + if (parseOptionalPositiveInt(sourceConfig.maxFiles) === null) { + return { valid: false, error: 'Max Files must be a positive whole number' } + } + + const unsupported = parseMultiValue(sourceConfig.extensions) + .map(normalizeExt) + .filter((ext) => ext && !isSupportedFileType(ext)) + if (unsupported.length > 0) { + return { + valid: false, + error: `Sim cannot extract text from these file types: ${unsupported.join(', ')}`, + } + } + + /** + * Existence check only — a foreign folder id leaks nothing, because every listing + * query is bound to the engine-supplied workspace regardless. Without it a typo'd + * id would just sync zero files forever with no explanation. + */ + const folderId = typeof sourceConfig.folderId === 'string' ? sourceConfig.folderId.trim() : '' + if (folderId && context?.workspaceId) { + const folders = await listWorkspaceFileFolders(context.workspaceId, { scope: 'active' }) + if (!folders.some((folder) => folder.id === folderId)) { + return { valid: false, error: 'Folder not found in this workspace' } + } + } + + return { valid: true } + }, + + mapTags: (metadata: Record): Record => { + const tags: Record = {} + + if (typeof metadata.folderPath === 'string' && metadata.folderPath) { + tags.folderPath = metadata.folderPath + } + if (typeof metadata.contentType === 'string' && metadata.contentType) { + tags.contentType = metadata.contentType + } + if (typeof metadata.uploadedBy === 'string' && metadata.uploadedBy) { + tags.uploadedBy = metadata.uploadedBy + } + if (typeof metadata.fileSize === 'number' && Number.isFinite(metadata.fileSize)) { + tags.fileSize = metadata.fileSize + } + const lastModified = parseTagDate(metadata.lastModified) + if (lastModified) tags.lastModified = lastModified + + return tags + }, +} diff --git a/apps/sim/connectors/types.ts b/apps/sim/connectors/types.ts index 71ad9ad6926..ccb5df067d4 100644 --- a/apps/sim/connectors/types.ts +++ b/apps/sim/connectors/types.ts @@ -5,10 +5,73 @@ import type { SelectorKey } from '@/hooks/selectors/types' * Authentication configuration for a connector. * OAuth connectors reuse the existing credential system. * API key connectors store an encrypted key in the `encryptedApiKey` column. + * Sim connectors read this workspace's own data and store no credential at all. */ export type ConnectorAuthConfig = | { mode: 'oauth'; provider: OAuthService; requiredScopes?: string[] } | { mode: 'apiKey'; label?: string; placeholder?: string } + /** + * Reads Sim's own data for the knowledge base's workspace. No credential is + * stored or resolved: authorization comes from the workspace permission the + * creator already had (`checkKnowledgeBaseWriteAccess`), and scope comes from + * {@link ConnectorSyncContext.workspaceId}, which the sync engine derives from + * the `knowledge_base` row and never from `sourceConfig`. + * + * A workspace API key would add nothing here: its only informational content is + * "which workspace", which the connector's own knowledge base already answers. + * + * The add-connector modal renders no auth row at all for this mode — the + * connector's own `description` already says it reads this workspace, and + * `sim-ui-copy` rules out restating that as helper text. + */ + | { mode: 'sim' } + +/** + * Whether the add-connector UI must collect a credential for this auth mode. + * Single source of truth for both the modal's auth field and the selector + * field's readiness gate, so the two can never disagree. + */ +export function collectsCredential(auth: ConnectorAuthConfig): boolean { + switch (auth.mode) { + case 'sim': + return false + case 'apiKey': + case 'oauth': + return true + default: { + const _exhaustive: never = auth + return true + } + } +} + +/** + * Per-run state shared across every {@link ConnectorConfig.listDocuments} and + * {@link ConnectorConfig.getDocument} call of a single sync. + * + * Extends an index signature so connectors can keep stashing ad-hoc caches on it + * (`syncContext.allFiles`, schema lookups) without declaring them here. + * + * The engine-supplied fields are `readonly`: a connector must not be able to + * reassign the workspace it is reading from. For `sim`-mode connectors this is + * the entire tenancy boundary, so it is expressed in the type system rather than + * left to a convention. + */ +export interface ConnectorSyncContext extends Record { + /** Identifies this sync run in logs. */ + readonly syncRunId: string + /** Derived from the `knowledge_base` row. Never read from `sourceConfig`. */ + readonly workspaceId: string + readonly knowledgeBaseId: string + /** + * Set by a connector when it could not enumerate the full source (a doc cap was + * hit, a scope lookup failed). Suppresses deletion reconciliation, because the + * engine hard-deletes anything absent from a listing it believes is complete. + */ + listingCapped?: boolean + /** Set by the engine when pagination stopped at `MAX_PAGES`. */ + listingTruncated?: boolean +} /** * A single document fetched from an external source. @@ -172,27 +235,35 @@ export interface ConnectorConfig extends ConnectorMeta { listDocuments: ( accessToken: string, sourceConfig: Record, - cursor?: string, - syncContext?: Record, + cursor: string | undefined, + syncContext: ConnectorSyncContext, lastSyncAt?: Date ) => Promise /** * Fetch a single document by its external ID. - * syncContext is an optional mutable object for caching expensive lookups - * (e.g. tag maps, notebook lists) across multiple getDocument calls. + * syncContext is the same mutable object listDocuments received, so caches + * (tag maps, notebook lists) carry across both phases of the run. */ getDocument: ( accessToken: string, sourceConfig: Record, externalId: string, - syncContext?: Record + syncContext: ConnectorSyncContext ) => Promise - /** Validate that sourceConfig is correct and accessible (called on save) */ + /** + * Validate that sourceConfig is correct and accessible (called on save). + * + * `context` carries the owning knowledge base's identifiers, resolved + * server-side from the KB row rather than from the submitted config. It is + * optional so existing connectors compile unchanged; `sim`-mode connectors use + * it to check that a referenced resource actually lives in this workspace. + */ validateConfig: ( accessToken: string, - sourceConfig: Record + sourceConfig: Record, + context?: { workspaceId?: string; knowledgeBaseId?: string } ) => Promise<{ valid: boolean; error?: string }> /** Map source metadata to semantic tag keys (translated to slots by the sync engine) */ diff --git a/apps/sim/hooks/queries/workspace-file-folders.ts b/apps/sim/hooks/queries/workspace-file-folders.ts index 40010ab03c8..a6a05b1c157 100644 --- a/apps/sim/hooks/queries/workspace-file-folders.ts +++ b/apps/sim/hooks/queries/workspace-file-folders.ts @@ -1,6 +1,12 @@ import { toast } from '@sim/emcn' import { toError } from '@sim/utils/errors' -import { keepPreviousData, useMutation, useQuery, useQueryClient } from '@tanstack/react-query' +import { + keepPreviousData, + type QueryFunctionContext, + useMutation, + useQuery, + useQueryClient, +} from '@tanstack/react-query' import { requestJson } from '@/lib/api/client/request' import { bulkArchiveWorkspaceFileItemsContract, @@ -49,16 +55,31 @@ export function invalidateWorkspaceFileBrowsers( queryClient.invalidateQueries({ queryKey: workspaceFilesKeys.storageInfo() }) } +/** + * Shared query options so non-hook consumers (the `sim.fileFolders` selector) read + * the exact cache entry the Files browser warms, rather than issuing a duplicate + * fetch under a parallel key. + */ +export function getWorkspaceFileFoldersQueryOptions( + workspaceId: string, + scope: WorkspaceFileFolderScope = 'active' +) { + return { + queryKey: workspaceFileFolderKeys.list(workspaceId, scope), + queryFn: ({ signal }: QueryFunctionContext) => + fetchWorkspaceFileFolders(workspaceId, scope, signal), + staleTime: WORKSPACE_FILE_FOLDERS_STALE_TIME, + } +} + export function useWorkspaceFileFolders( workspaceId: string, scope: WorkspaceFileFolderScope = 'active', options?: { enabled?: boolean } ) { return useQuery({ - queryKey: workspaceFileFolderKeys.list(workspaceId, scope), - queryFn: ({ signal }) => fetchWorkspaceFileFolders(workspaceId, scope, signal), + ...getWorkspaceFileFoldersQueryOptions(workspaceId, scope), enabled: Boolean(workspaceId) && (options?.enabled ?? true), - staleTime: WORKSPACE_FILE_FOLDERS_STALE_TIME, placeholderData: keepPreviousData, }) } diff --git a/apps/sim/hooks/selectors/providers/sim/selectors.ts b/apps/sim/hooks/selectors/providers/sim/selectors.ts index 13af6d6d08e..f2d31359530 100644 --- a/apps/sim/hooks/selectors/providers/sim/selectors.ts +++ b/apps/sim/hooks/selectors/providers/sim/selectors.ts @@ -5,6 +5,10 @@ import { getFolderMap } from '@/hooks/queries/utils/folder-cache' import { collectDuplicateNames, disambiguateLabelByFolder } from '@/hooks/queries/utils/folder-tree' import { getWorkflowById, getWorkflows } from '@/hooks/queries/utils/workflow-cache' import { getWorkflowListQueryOptions } from '@/hooks/queries/utils/workflow-list-query' +import { + getWorkspaceFileFoldersQueryOptions, + type WorkspaceFileFolderApi, +} from '@/hooks/queries/workspace-file-folders' import { SELECTOR_STALE } from '@/hooks/selectors/providers/shared' import { selectorKeys } from '@/hooks/selectors/query-keys' import type { @@ -15,6 +19,16 @@ import type { } from '@/hooks/selectors/types' import type { WorkflowMetadata } from '@/stores/workflows/registry/types' +/** + * The API already returns a slash-joined ancestor path, so spacing it out is the + * whole nesting affordance — and sorting by the rendered label groups children + * directly under their parent. + */ +function folderPathLabel(folder: WorkspaceFileFolderApi): string { + const path = folder.path.trim() + return path ? path.split('/').join(' / ') : folder.name +} + /** Matches the workflow list's own fallback for an unnamed workflow. */ function workflowBaseLabel(workflow: WorkflowMetadata): string { return workflow.name || `Workflow ${workflow.id.slice(0, 8)}` @@ -71,6 +85,32 @@ export const simSelectors = { } }, }, + 'sim.fileFolders': { + key: 'sim.fileFolders', + staleTime: SELECTOR_STALE, + getQueryKey: ({ context }: SelectorQueryArgs) => + context.workspaceId + ? selectorKeys.simFileFolders(context.workspaceId) + : [...selectorKeys.all, 'sim.fileFolders', 'none'], + enabled: ({ context }) => Boolean(context.workspaceId), + fetchList: async ({ context }: SelectorQueryArgs): Promise => { + if (!context.workspaceId) return [] + const folders = await getQueryClient().ensureQueryData( + getWorkspaceFileFoldersQueryOptions(context.workspaceId) + ) + return folders + .map((folder) => ({ id: folder.id, label: folderPathLabel(folder) })) + .sort((a, b) => a.label.localeCompare(b.label)) + }, + fetchById: async ({ context, detailId }: SelectorQueryArgs): Promise => { + if (!detailId || !context.workspaceId) return null + const folders = await getQueryClient().ensureQueryData( + getWorkspaceFileFoldersQueryOptions(context.workspaceId) + ) + const folder = folders.find((f) => f.id === detailId) + return folder ? { id: folder.id, label: folderPathLabel(folder) } : null + }, + }, 'table.columns': { key: 'table.columns', staleTime: SELECTOR_STALE, @@ -100,4 +140,7 @@ export const simSelectors = { return col ? { id: getColumnId(col), label: col.name } : null }, }, -} satisfies Record, SelectorDefinition> +} satisfies Record< + Extract, + SelectorDefinition +> diff --git a/apps/sim/hooks/selectors/query-keys.ts b/apps/sim/hooks/selectors/query-keys.ts index c5fa1afe97a..a4223f7e280 100644 --- a/apps/sim/hooks/selectors/query-keys.ts +++ b/apps/sim/hooks/selectors/query-keys.ts @@ -4,4 +4,6 @@ export const selectorKeys = { [...selectorKeys.all, 'sim.workflows', workspaceId] as const, simWorkflows: (workspaceId: string, excludeWorkflowId?: string) => [...selectorKeys.simWorkflowsPrefix(workspaceId), excludeWorkflowId ?? 'none'] as const, + simFileFolders: (workspaceId: string) => + [...selectorKeys.all, 'sim.fileFolders', workspaceId] as const, } diff --git a/apps/sim/hooks/selectors/registry.test.ts b/apps/sim/hooks/selectors/registry.test.ts index 7114545606a..1fdd87bccb6 100644 --- a/apps/sim/hooks/selectors/registry.test.ts +++ b/apps/sim/hooks/selectors/registry.test.ts @@ -7,6 +7,7 @@ import * as getQueryClientModule from '@/app/_shell/providers/get-query-client' import * as folderCacheModule from '@/hooks/queries/utils/folder-cache' import * as workflowCacheModule from '@/hooks/queries/utils/workflow-cache' import * as workflowListQueryModule from '@/hooks/queries/utils/workflow-list-query' +import * as workspaceFileFoldersModule from '@/hooks/queries/workspace-file-folders' import { getSelectorDefinition } from '@/hooks/selectors/registry' const mockEnsureQueryData = vi.fn().mockResolvedValue(undefined) @@ -36,7 +37,19 @@ const getWorkflowListQueryOptionsSpy = vi }) as unknown as ReturnType ) +const getWorkspaceFileFoldersQueryOptionsSpy = vi + .spyOn(workspaceFileFoldersModule, 'getWorkspaceFileFoldersQueryOptions') + .mockImplementation( + (workspaceId: string) => + ({ + queryKey: ['workspaceFileFolders', 'list', workspaceId, 'active'], + }) as unknown as ReturnType< + typeof workspaceFileFoldersModule.getWorkspaceFileFoldersQueryOptions + > + ) + afterAll(() => { + getWorkspaceFileFoldersQueryOptionsSpy.mockRestore() getQueryClientSpy.mockRestore() mockGetWorkflows.mockRestore() getWorkflowByIdSpy.mockRestore() @@ -170,3 +183,82 @@ describe('sim.workflows selector', () => { expect(option).toEqual({ id: 'wf-1', label: 'Pipeline (Alpha)' }) }) }) + +describe('sim.fileFolders selector', () => { + const FOLDERS = [ + { id: 'f-specs', name: 'Specs', path: 'Docs/Specs' }, + { id: 'f-docs', name: 'Docs', path: 'Docs' }, + { id: 'f-root', name: 'Loose', path: '' }, + ] + + beforeEach(() => { + vi.clearAllMocks() + mockEnsureQueryData.mockResolvedValue(FOLDERS) + getQueryClientSpy.mockImplementation( + () => ({ ensureQueryData: mockEnsureQueryData }) as unknown as QueryClient + ) + getWorkspaceFileFoldersQueryOptionsSpy.mockImplementation( + (workspaceId: string) => + ({ + queryKey: ['workspaceFileFolders', 'list', workspaceId, 'active'], + }) as unknown as ReturnType< + typeof workspaceFileFoldersModule.getWorkspaceFileFoldersQueryOptions + > + ) + }) + + /** + * Guards the failure mode where the key is added to the `SelectorKey` union but the + * `satisfies Extract<...>` clause is not widened: the selector then type-checks + * everywhere yet is absent from the registry, and only throws once a field renders. + */ + it('is registered and workspace-scoped', () => { + const definition = getSelectorDefinition('sim.fileFolders') + + expect(definition.enabled?.({ key: 'sim.fileFolders', context: {} })).toBe(false) + expect(definition.enabled?.({ key: 'sim.fileFolders', context: { workspaceId: 'ws-1' } })).toBe( + true + ) + expect( + definition.getQueryKey({ key: 'sim.fileFolders', context: { workspaceId: 'ws-1' } }) + ).toEqual(['selectors', 'sim.fileFolders', 'ws-1']) + }) + + it('renders nested folders as spaced paths sorted so children follow their parent', async () => { + const definition = getSelectorDefinition('sim.fileFolders') + + const options = await definition.fetchList!({ + key: 'sim.fileFolders', + context: { workspaceId: 'ws-1' }, + }) + + expect(mockEnsureQueryData).toHaveBeenCalledWith({ + queryKey: ['workspaceFileFolders', 'list', 'ws-1', 'active'], + }) + expect(options).toEqual([ + { id: 'f-docs', label: 'Docs' }, + { id: 'f-specs', label: 'Docs / Specs' }, + { id: 'f-root', label: 'Loose' }, + ]) + }) + + it('labels a saved folder id so reopening the modal shows a name, not a uuid', async () => { + const definition = getSelectorDefinition('sim.fileFolders') + + await expect( + definition.fetchById?.({ + key: 'sim.fileFolders', + context: { workspaceId: 'ws-1' }, + detailId: 'f-specs', + }) + ).resolves.toEqual({ id: 'f-specs', label: 'Docs / Specs' }) + + await expect( + definition.fetchById?.({ + key: 'sim.fileFolders', + context: { workspaceId: 'ws-1' }, + detailId: 'deleted-folder', + }) + ).resolves.toBeNull() + }) +}) diff --git a/apps/sim/hooks/selectors/types.ts b/apps/sim/hooks/selectors/types.ts index 88ea585d797..f3332cf614d 100644 --- a/apps/sim/hooks/selectors/types.ts +++ b/apps/sim/hooks/selectors/types.ts @@ -64,6 +64,7 @@ export type SelectorKey = | 'monday.boards' | 'monday.groups' | 'sim.workflows' + | 'sim.fileFolders' | 'table.columns' export interface SelectorOption { diff --git a/apps/sim/lib/knowledge/connectors/source-config.test.ts b/apps/sim/lib/knowledge/connectors/source-config.test.ts new file mode 100644 index 00000000000..5f190ea1787 --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/source-config.test.ts @@ -0,0 +1,85 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { + preserveServerOwnedSourceConfig, + RESERVED_SOURCE_CONFIG_KEYS, + sanitizeConnectorSourceConfig, +} from '@/lib/knowledge/connectors/source-config' + +describe('sanitizeConnectorSourceConfig', () => { + /** + * The tenancy control for `sim`-mode connectors: the engine derives the workspace + * from the knowledge_base row, so a caller-supplied one must never be persisted + * where a connector could read it back. + */ + it('strips every reserved key a caller could use to widen scope', () => { + expect( + sanitizeConnectorSourceConfig({ + workspaceId: 'victim-ws', + knowledgeBaseId: 'victim-kb', + tagSlotMapping: { folderPath: 'tag7' }, + folderId: 'f-1', + recursive: 'false', + }) + ).toEqual({ folderId: 'f-1', recursive: 'false' }) + }) + + it('covers the whole declared reserved list', () => { + const everyReserved = Object.fromEntries(RESERVED_SOURCE_CONFIG_KEYS.map((k) => [k, 'x'])) + expect(sanitizeConnectorSourceConfig(everyReserved)).toEqual({}) + }) + + it('leaves unreserved keys untouched, including falsy values', () => { + const input = { folderId: '', recursive: 'false', maxFiles: 0 } + expect(sanitizeConnectorSourceConfig(input)).toEqual(input) + }) + + it('does not mutate the caller object', () => { + const input = { workspaceId: 'victim-ws', folderId: 'f-1' } + sanitizeConnectorSourceConfig(input) + expect(input.workspaceId).toBe('victim-ws') + }) +}) + +describe('preserveServerOwnedSourceConfig', () => { + /** + * Update replaces `sourceConfig` wholesale. Without this, sanitizing would drop + * `tagSlotMapping` on every edit and the connector would silently stop writing + * tags — for every connector that declares tagDefinitions, not just the sim ones. + */ + it('carries the stored tagSlotMapping across an edit that does not resend it', () => { + expect( + preserveServerOwnedSourceConfig( + { folderId: 'new-folder' }, + { folderId: 'old-folder', tagSlotMapping: { folderPath: 'tag1' } } + ) + ).toEqual({ folderId: 'new-folder', tagSlotMapping: { folderPath: 'tag1' } }) + }) + + /** The stored mapping wins: a caller cannot claim slots it was not allocated. */ + it('prefers the stored mapping over anything left in the update', () => { + const result = preserveServerOwnedSourceConfig( + { tagSlotMapping: { folderPath: 'tag7' } } as Record, + { tagSlotMapping: { folderPath: 'tag1' } } + ) + expect(result.tagSlotMapping).toEqual({ folderPath: 'tag1' }) + }) + + /** workspaceId/knowledgeBaseId are never persisted, so nothing should resurrect them. */ + it('does not resurrect keys that are never persisted', () => { + const result = preserveServerOwnedSourceConfig( + { folderId: 'f-1' }, + { workspaceId: 'victim-ws', knowledgeBaseId: 'victim-kb' } + ) + expect(result).toEqual({ folderId: 'f-1' }) + }) + + it('tolerates a connector row with no stored config', () => { + expect(preserveServerOwnedSourceConfig({ folderId: 'f-1' }, null)).toEqual({ folderId: 'f-1' }) + expect(preserveServerOwnedSourceConfig({ folderId: 'f-1' }, undefined)).toEqual({ + folderId: 'f-1', + }) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/source-config.ts b/apps/sim/lib/knowledge/connectors/source-config.ts new file mode 100644 index 00000000000..6ab266e2c91 --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/source-config.ts @@ -0,0 +1,60 @@ +import { omit } from '@sim/utils/object' + +/** + * Keys a caller must never be able to persist into `knowledge_connector.sourceConfig`. + * + * `workspaceId` and `knowledgeBaseId` are supplied by the sync engine at run time from + * the `knowledge_base` row (see `ConnectorSyncContext`). For `sim`-mode connectors that + * binding is the entire tenancy boundary, so allowing a stored copy would create a + * second, caller-controlled source of truth for which workspace to read. + * + * `tagSlotMapping` is derived during connector creation from the knowledge base's + * available tag slots; a caller-supplied one would let a connector write into slots it + * was never allocated. + */ +export const RESERVED_SOURCE_CONFIG_KEYS = [ + 'workspaceId', + 'knowledgeBaseId', + 'tagSlotMapping', +] as const + +/** + * Strips {@link RESERVED_SOURCE_CONFIG_KEYS} from a caller-submitted source config. + * Applied on both create and update so the two paths cannot drift. + */ +export function sanitizeConnectorSourceConfig( + sourceConfig: Record +): Record { + return omit(sourceConfig, [...RESERVED_SOURCE_CONFIG_KEYS]) +} + +/** + * The reserved keys that are legitimately *persisted*, just never by the caller. + * + * `workspaceId` and `knowledgeBaseId` are stripped and never stored at all — the + * engine derives them per run. `tagSlotMapping` is different: it is computed once + * during creation from the knowledge base's free slots and must survive edits. + */ +const SERVER_OWNED_PERSISTED_KEYS = ['tagSlotMapping'] as const + +/** + * Re-applies server-owned keys from the stored row onto a sanitized update. + * + * Update replaces `sourceConfig` wholesale, so sanitizing alone would drop + * `tagSlotMapping` — which the client never re-sends. Losing it makes + * `resolveTagMapping` return undefined and the connector silently stops writing + * tags on every later sync, for every connector that declares `tagDefinitions`. + * Reading it back from the stored row rather than the request keeps the key + * server-owned while still surviving an edit. + */ +export function preserveServerOwnedSourceConfig( + sanitizedUpdate: Record, + storedSourceConfig: unknown +): Record { + const stored = (storedSourceConfig ?? {}) as Record + const preserved: Record = {} + for (const key of SERVER_OWNED_PERSISTED_KEYS) { + if (stored[key] !== undefined) preserved[key] = stored[key] + } + return { ...sanitizedUpdate, ...preserved } +} diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index 540ae694595..c17334b802f 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -1,7 +1,7 @@ /** * @vitest-environment node */ -import { authOAuthUtilsMock } from '@sim/testing' +import { authOAuthUtilsMock, authOAuthUtilsMockFns } from '@sim/testing' import { generateShortId } from '@sim/utils/id' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -19,6 +19,15 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ })) vi.mock('@/lib/uploads', () => ({ StorageService: {} })) vi.mock('@/app/api/auth/oauth/utils', () => authOAuthUtilsMock) + +const { mockResolveCredentialTokenIdentity, mockDecryptApiKey } = vi.hoisted(() => ({ + mockResolveCredentialTokenIdentity: vi.fn(), + mockDecryptApiKey: vi.fn(), +})) +vi.mock('@/lib/credentials/access', () => ({ + resolveCredentialTokenIdentity: mockResolveCredentialTokenIdentity, +})) +vi.mock('@/lib/api-key/crypto', () => ({ decryptApiKey: mockDecryptApiKey })) vi.mock('@/background/knowledge-connector-sync', () => ({ knowledgeConnectorSync: { trigger: vi.fn() }, })) @@ -537,3 +546,144 @@ describe('chunkOpsByByteBudget', () => { expect(chunks).toHaveLength(1) }) }) + +/** + * `resolveConnectorAuth` is on the credential path for every connector, not just the + * `sim` ones — it absorbed the credential-identity lookup that used to sit inline in + * `executeSync`, including the service-account-vs-oauth choice of WHICH user's token + * to read. That swap is the regression this refactor is most likely to reintroduce. + */ +describe('resolveConnectorAuth', () => { + const owner = { workspaceId: 'ws-1', userId: 'kb-owner' } + const noCredential = { credentialId: null, encryptedApiKey: null } + + beforeEach(() => { + vi.clearAllMocks() + }) + + it('resolves sim mode without touching any credential system', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + + await expect( + resolveConnectorAuth(noCredential, { auth: { mode: 'sim' } }, owner) + ).resolves.toEqual({ mode: 'sim' }) + expect(mockResolveCredentialTokenIdentity).not.toHaveBeenCalled() + expect(authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded).not.toHaveBeenCalled() + expect(mockDecryptApiKey).not.toHaveBeenCalled() + }) + + /** The `sim` arm must carry no token field at all, or the empty-bearer sentinel returns. */ + it('exposes no accessToken for sim mode', async () => { + const { resolveConnectorAuth, tokenFor } = await import( + '@/lib/knowledge/connectors/sync-engine' + ) + + const resolved = await resolveConnectorAuth(noCredential, { auth: { mode: 'sim' } }, owner) + expect(resolved).not.toHaveProperty('accessToken') + expect(tokenFor(resolved)).toBe('') + }) + + it('decrypts the stored key for apiKey mode', async () => { + const { resolveConnectorAuth, tokenFor } = await import( + '@/lib/knowledge/connectors/sync-engine' + ) + mockDecryptApiKey.mockResolvedValue({ decrypted: 'secret-key' }) + + const resolved = await resolveConnectorAuth( + { credentialId: null, encryptedApiKey: 'enc' }, + { auth: { mode: 'apiKey' } }, + owner + ) + expect(resolved).toEqual({ mode: 'apiKey', accessToken: 'secret-key' }) + expect(tokenFor(resolved)).toBe('secret-key') + }) + + it('throws when an apiKey connector has no stored key', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + + await expect( + resolveConnectorAuth(noCredential, { auth: { mode: 'apiKey' } }, owner) + ).rejects.toThrow(/missing encrypted API key/) + }) + + it('throws when an oauth connector has no credential id', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + + await expect( + resolveConnectorAuth(noCredential, { auth: { mode: 'oauth', provider: 'jira' } }, owner) + ).rejects.toThrow(/missing credential ID/) + }) + + /** + * Workspace credentials are routinely authorized by someone other than the KB + * owner, and token reads are scoped to `account.userId` — reading as the KB owner + * resolves no token at all. + */ + it('reads the token as the credential owner, not the knowledge base owner', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + mockResolveCredentialTokenIdentity.mockResolvedValue({ + kind: 'oauth', + userId: 'credential-owner', + }) + authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded.mockResolvedValue('tok') + + const resolved = await resolveConnectorAuth( + { credentialId: 'cred-1', encryptedApiKey: null }, + { auth: { mode: 'oauth', provider: 'jira' } }, + owner + ) + + expect(authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded).toHaveBeenCalledWith( + 'cred-1', + 'credential-owner', + expect.any(String) + ) + expect(resolved).toMatchObject({ mode: 'oauth', credentialUserId: 'credential-owner' }) + }) + + /** Service accounts mint their own token and ignore the acting user. */ + it('falls back to the acting user for a service-account identity', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + mockResolveCredentialTokenIdentity.mockResolvedValue({ kind: 'service_account' }) + authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded.mockResolvedValue('tok') + + await resolveConnectorAuth( + { credentialId: 'cred-1', encryptedApiKey: null }, + { auth: { mode: 'oauth', provider: 'jira' } }, + owner + ) + + expect(authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded).toHaveBeenCalledWith( + 'cred-1', + 'kb-owner', + expect.any(String) + ) + }) + + it('throws when the credential is not usable from the workspace', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + mockResolveCredentialTokenIdentity.mockResolvedValue(null) + + await expect( + resolveConnectorAuth( + { credentialId: 'cred-1', encryptedApiKey: null }, + { auth: { mode: 'oauth', provider: 'jira' } }, + owner + ) + ).rejects.toThrow(/not usable from workspace/) + }) + + it('throws when the refresh yields no token', async () => { + const { resolveConnectorAuth } = await import('@/lib/knowledge/connectors/sync-engine') + mockResolveCredentialTokenIdentity.mockResolvedValue({ kind: 'oauth', userId: 'u' }) + authOAuthUtilsMockFns.mockRefreshAccessTokenIfNeeded.mockResolvedValue(null) + + await expect( + resolveConnectorAuth( + { credentialId: 'cred-1', encryptedApiKey: null }, + { auth: { mode: 'oauth', provider: 'jira' } }, + owner + ) + ).rejects.toThrow(/Failed to obtain access token/) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index 8ca978ec829..a02dcc6be8e 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -28,6 +28,7 @@ import { refreshAccessTokenIfNeeded } from '@/app/api/auth/oauth/utils' import { CONNECTOR_REGISTRY } from '@/connectors/registry.server' import type { ConnectorAuthConfig, + ConnectorSyncContext, DocumentTags, ExternalDocument, SyncResult, @@ -375,47 +376,106 @@ export function resolveTagMapping( } /** - * Resolves an access token for a connector based on its auth mode. - * OAuth connectors refresh via the credential system; API key connectors - * decrypt the key stored in the dedicated `encryptedApiKey` column. + * A connector's resolved credential for one sync run. * - * `userId` must be the user who owns the credential's OAuth account — not the - * knowledge base owner. Workspace-scoped credentials are routinely authorized by - * a different member, and token reads are scoped to `account.userId`. + * The `sim` arm carries no `accessToken` at all, so a credential-less connector + * cannot be handed an empty-string bearer by accident, and no caller can guard on + * a token's falsiness. Callers narrow on `mode` instead. */ -async function resolveAccessToken( +export type ResolvedConnectorAuth = + | { mode: 'oauth'; accessToken: string; credentialUserId: string } + | { mode: 'apiKey'; accessToken: string } + | { mode: 'sim' } + +/** + * The value handed to a connector's `listDocuments`/`getDocument` as their first + * argument. Those signatures are shared by every connector, so `sim` connectors — + * which ignore the argument entirely and read the database directly — receive an + * empty string here. This is the only place the empty token exists; it never + * reaches control flow. + */ +export function tokenFor(auth: ResolvedConnectorAuth): string { + return auth.mode === 'sim' ? '' : auth.accessToken +} + +/** + * Resolves a connector's credential based on its auth mode. + * OAuth connectors refresh via the credential system; API key connectors decrypt + * the key stored in the dedicated `encryptedApiKey` column; `sim` connectors have + * no credential to resolve and skip both. + * + * For OAuth this also resolves *whose* account the token is read from: workspace + * credentials are routinely authorized by a member who is not the knowledge base + * owner, and token reads are scoped to `account.userId`, so passing the KB owner + * resolves no token at all. Identity and token are resolved together so they can + * never disagree. + */ +export async function resolveConnectorAuth( connector: { credentialId: string | null; encryptedApiKey: string | null }, connectorConfig: { auth: ConnectorAuthConfig }, - userId: string -): Promise { - if (connectorConfig.auth.mode === 'apiKey') { - if (!connector.encryptedApiKey) { - throw new Error('API key connector is missing encrypted API key') + /** + * Non-null `workspaceId` is a precondition, not a convenience: it scopes the + * OAuth credential lookup and is the entire tenancy boundary for `sim` mode. + * `executeSync` asserts it before this call. + */ + owner: { workspaceId: string; userId: string } +): Promise { + switch (connectorConfig.auth.mode) { + case 'sim': + return { mode: 'sim' } + + case 'apiKey': { + if (!connector.encryptedApiKey) { + throw new Error('API key connector is missing encrypted API key') + } + const { decrypted } = await decryptApiKey(connector.encryptedApiKey) + return { mode: 'apiKey', accessToken: decrypted } } - const { decrypted } = await decryptApiKey(connector.encryptedApiKey) - return decrypted - } - if (!connector.credentialId) { - throw new Error('OAuth connector is missing credential ID') - } + case 'oauth': { + if (!connector.credentialId) { + throw new Error('OAuth connector is missing credential ID') + } - const requestId = `sync-${connector.credentialId}` - const token = await refreshAccessTokenIfNeeded(connector.credentialId, userId, requestId) + const identity = await resolveCredentialTokenIdentity( + connector.credentialId, + owner.workspaceId + ) + if (!identity) { + throw new Error( + `Credential ${connector.credentialId} is not usable from workspace ${owner.workspaceId} — reconnect the credential` + ) + } + // Service accounts mint their own token and ignore the acting user. + const credentialUserId = identity.kind === 'oauth' ? identity.userId : owner.userId - if (!token) { - logger.error(`[${requestId}] refreshAccessTokenIfNeeded returned null`, { - credentialId: connector.credentialId, - userId, - authMode: connectorConfig.auth.mode, - authProvider: connectorConfig.auth.provider, - }) - throw new Error( - `Failed to obtain access token for credential ${connector.credentialId} (provider: ${connectorConfig.auth.provider})` - ) - } + const requestId = `sync-${connector.credentialId}` + const token = await refreshAccessTokenIfNeeded( + connector.credentialId, + credentialUserId, + requestId + ) - return token + if (!token) { + logger.error(`[${requestId}] refreshAccessTokenIfNeeded returned null`, { + credentialId: connector.credentialId, + userId: credentialUserId, + authMode: connectorConfig.auth.mode, + authProvider: connectorConfig.auth.provider, + }) + throw new Error( + `Failed to obtain access token for credential ${connector.credentialId} (provider: ${connectorConfig.auth.provider})` + ) + } + + return { mode: 'oauth', accessToken: token, credentialUserId } + } + + default: { + const _exhaustive: never = connectorConfig.auth + throw new Error(`Unsupported connector auth mode: ${JSON.stringify(_exhaustive)}`) + } + } } /** @@ -489,14 +549,17 @@ export async function executeSync( } const userId = kbRows[0].userId - // Resolved once per sync and threaded into add/updateDocument so every synced - // kb/ object records a trusted ownership binding without an N+1 KB lookup. - const kbOwner: KnowledgeBaseOwner = { workspaceId: kbRows[0].workspaceId, userId } - if (!kbOwner.workspaceId) { + // Narrowed before `kbOwner` is built so the non-null value can flow into + // `resolveConnectorAuth` and `syncContext`, both of which require a real workspace. + const workspaceId = kbRows[0].workspaceId + if (!workspaceId) { throw new Error( `Knowledge base ${connector.knowledgeBaseId} is missing workspace billing context` ) } + // Resolved once per sync and threaded into add/updateDocument so every synced + // kb/ object records a trusted ownership binding without an N+1 KB lookup. + const kbOwner: KnowledgeBaseOwner = { workspaceId, userId } if (billingAttribution.workspaceId !== kbOwner.workspaceId) { throw new Error( `Connector sync billing attribution does not match knowledge base workspace ${kbOwner.workspaceId}` @@ -534,36 +597,25 @@ export async function executeSync( let syncExitedCleanly = false try { - /** - * OAuth credentials are workspace-scoped and shared, so the member who authorized - * one is often not the knowledge base owner. Resolve the credential's own account - * owner — token reads are scoped to `account.userId`, so passing the KB owner - * resolves no token at all. Resolved once here rather than inside - * `resolveAccessToken` so per-page refreshes don't repeat the lookup. - */ - let credentialUserId = userId - if (connectorConfig.auth.mode === 'oauth' && connector.credentialId) { - const identity = await resolveCredentialTokenIdentity( - connector.credentialId, - kbOwner.workspaceId - ) - if (!identity) { - throw new Error( - `Credential ${connector.credentialId} is not usable from workspace ${kbOwner.workspaceId} — reconnect the credential` - ) - } - // Service accounts mint their own token and ignore the acting user. - if (identity.kind === 'oauth') { - credentialUserId = identity.userId - } - } - - let accessToken = await resolveAccessToken(connector, connectorConfig, credentialUserId) + let resolvedAuth = await resolveConnectorAuth(connector, connectorConfig, { + workspaceId, + userId, + }) const externalDocs: ExternalDocument[] = [] let cursor: string | undefined let hasMore = true - const syncContext: Record = { syncRunId: generateId() } + /** + * `workspaceId` and `knowledgeBaseId` come from the `knowledge_base` row read + * above, never from `sourceConfig`. For `sim`-mode connectors this is the whole + * tenancy boundary, and `ConnectorSyncContext` declares them `readonly` so a + * connector cannot widen its own scope. + */ + const syncContext: ConnectorSyncContext = { + syncRunId: generateId(), + workspaceId, + knowledgeBaseId: connector.knowledgeBaseId, + } // Shared cutoff for both the tombstone-retry bound below and the stuck-document // retry near the end of this sync — same RETRY_WINDOW_DAYS window, one computation. @@ -633,12 +685,17 @@ export async function executeSync( ) for (let pageNum = 0; hasMore && pageNum < MAX_PAGES; pageNum++) { - if (pageNum > 0 && connectorConfig.auth.mode === 'oauth') { - accessToken = await resolveAccessToken(connector, connectorConfig, credentialUserId) + // Only OAuth access tokens expire mid-run; API keys and `sim` connectors + // hold for the whole sync. + if (pageNum > 0 && resolvedAuth.mode === 'oauth') { + resolvedAuth = await resolveConnectorAuth(connector, connectorConfig, { + workspaceId, + userId, + }) } const page = await connectorConfig.listDocuments( - accessToken, + tokenFor(resolvedAuth), sourceConfig, cursor, syncContext, @@ -823,14 +880,20 @@ export async function executeSync( const readyOps = contentOps.filter((op) => !op.extDoc.contentDeferred) if (deferredOps.length > 0) { - if (connectorConfig.auth.mode === 'oauth') { - accessToken = await resolveAccessToken(connector, connectorConfig, credentialUserId) + // Hydration can start long after the listing finished, so refresh the one + // credential kind that expires. + if (resolvedAuth.mode === 'oauth') { + resolvedAuth = await resolveConnectorAuth(connector, connectorConfig, { + workspaceId, + userId, + }) } + const hydrationToken = tokenFor(resolvedAuth) const hydrated = await Promise.allSettled( deferredOps.map(async (op) => { const fullDoc = await connectorConfig.getDocument( - accessToken!, + hydrationToken, sourceConfig, op.extDoc.externalId, syncContext diff --git a/packages/testing/src/mocks/schema.mock.ts b/packages/testing/src/mocks/schema.mock.ts index ebd71000c61..d2057e13bbf 100644 --- a/packages/testing/src/mocks/schema.mock.ts +++ b/packages/testing/src/mocks/schema.mock.ts @@ -587,9 +587,12 @@ export const schemaMock = { key: 'key', userId: 'userId', workspaceId: 'workspaceId', + folderId: 'folderId', context: 'context', chatId: 'chatId', + messageId: 'messageId', originalName: 'originalName', + displayName: 'displayName', contentType: 'contentType', size: 'size', deletedAt: 'deletedAt',